diff --git a/migrations/inputs_sflow/migration_test.go b/migrations/inputs_sflow/migration_test.go index da7e8e2576d4f..5e686491a1eb4 100644 --- a/migrations/inputs_sflow/migration_test.go +++ b/migrations/inputs_sflow/migration_test.go @@ -10,7 +10,6 @@ import ( "github.com/influxdata/telegraf/config" _ "github.com/influxdata/telegraf/migrations/inputs_sflow" // register migration _ "github.com/influxdata/telegraf/plugins/inputs/netflow" // register plugin - _ "github.com/influxdata/telegraf/plugins/inputs/sflow" // register plugin ) func TestCases(t *testing.T) { diff --git a/plugins/inputs/all/sflow.go b/plugins/inputs/all/sflow.go deleted file mode 100644 index d75527cbe94d9..0000000000000 --- a/plugins/inputs/all/sflow.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !custom || inputs || inputs.sflow - -package all - -import _ "github.com/influxdata/telegraf/plugins/inputs/sflow" // register plugin diff --git a/plugins/inputs/sflow/README.md b/plugins/inputs/sflow/README.md deleted file mode 100644 index 56a00dd9b6cd2..0000000000000 --- a/plugins/inputs/sflow/README.md +++ /dev/null @@ -1,134 +0,0 @@ -# SFlow Input Plugin - -This service plugin produces metrics from information received by acting as a -[SFlow V5][sflow_v5] collector. Currently, the plugin can collect Flow Samples -of Ethernet / IPv4, IPv4 TCP and UDP headers. Counters and other header samples -are ignored. Please use the [netflow plugin][netflow] for a more modern and -sophisticated implementation. - -> [!CRITICAL] -> This plugin produces high cardinality data, which when not controlled for will -> cause high load on your database. Please make sure to [filter][filtering] the -> produced metrics or configure your database to avoid cardinality issues! - -⭐ Telegraf v1.14.0 -🏷️ network -💻 all - -[sflow_v5]: https://sflow.org/sflow_version_5.txt -[netflow]: /plugins/inputs/netflow/README.md -[filtering]: /docs/CONFIGURATION.md#metric-filtering - -## Service Input - -This plugin is a service input. Normal plugins gather metrics determined by the -interval setting. Service plugins start a service to listen and wait for -metrics or events to occur. Service plugins have two key differences from -normal plugins: - -1. The global or plugin specific `interval` setting may not apply -2. The CLI options of `--test`, `--test-wait`, and `--once` may not produce - output for this plugin - -## Global configuration options - -Plugins support additional global and plugin configuration settings for tasks -such as modifying metrics, tags, and fields, creating aliases, and configuring -plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details. - -[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins - -## Configuration - -```toml @sample.conf -# SFlow V5 Protocol Listener -[[inputs.sflow]] - ## Address to listen for sFlow packets. - ## example: service_address = "udp://:6343" - ## service_address = "udp4://:6343" - ## service_address = "udp6://:6343" - service_address = "udp://:6343" - - ## Set the size of the operating system's receive buffer. - ## example: read_buffer_size = "64KiB" - # read_buffer_size = "" -``` - -## Metrics - -- sflow - - tags: - - agent_address - IP address of the agent obtaining the sflow sample and - sent it to this collector - - source_id_type - type of flow_sample or flow_sample_expanded structures - - source_id_index - index of flow_sample or flow_sample_expanded structures - - input_ifindex - input value of flow_sample or flow_sample_expanded structures - - output_ifindex - output value of flow_sample or flow_sample_expanded structures - - sample_direction - source_id_index, netif_index_in and netif_index_out - - header_protocol - header_protocol field of sampled_header structures - - ether_type - ethernet type of an ETHERNET-ISO88023 header - - src_ip - source IP address of IPv4 or IPv6 structures - - src_port - source port of TCP or UDP structures - - src_port_name - name of the source port - - src_mac - source MAC address of an ETHERNET-ISO88023 header - - src_vlan - source VLAN of extended_switch structure - - src_priority - source priority of extended_switch structure - - src_mask_len - length of source mask of extended_router structure - - dst_ip - destination IP address of IPv4 or IPv6 structures - - dst_port - destination port of TCP or UDP structures - - dst_port_name - name of the destination port - - dst_mac - destination MAC address of an ETHERNET-ISO88023 header - - dst_vlan - destination VLAN of extended_switch structure - - dst_priority - destination priority extended_switch structure - - dst_mask_len - length of destinationd mask of extended_router structure - - next_hop - next hop of extended_router structure - - ip_version - IP version of IPv4 or IPv6 structures - - ip_protocol - IP protocol of IPv4 or IPv6 structures - - ip_dscp - IP DSCP of IPv4 or IPv6 structures - - ip_ecn - IP ECN of IPv4 or IPv6 structures - - tcp_urgent_pointer - urgent pointer of TCP structure - - fields: - - bytes (int) - product of frame length and packets - - drops (int) - drops field of flow_sample or - flow_sample_expanded structures - - packets (int) - sampling_rate field of flow_sample or - flow_sample_expanded structures - - frame_length (int) - frame_length field of sampled_header structures - - header_size (int) - header_size field of sampled_header structures - - ip_fragment_offset (int) - ip_ver field of IPv4 structures - - ip_header_length (int) - ip_ver field of IPv4 structures - - ip_total_length (int) - ip_total_len field of IPv4 structures - - ip_ttl (int) - ip_ttl field of IPv4 structures or - ip_hop_limit field IPv6 structures - - tcp_header_length (int) - size field of TCP structure. This value is - specified in 32-bit words. It must be multiplied - by 4 to produce a valuein bytes. - - tcp_window_size (int) - window_size field of TCP structure - - udp_length (int) - length field of UDP structures - - ip_flags (int) - ip_ver field of IPv4 structures - - tcp_flags (int) - TCP flags of TCP IP header (IPv4 or IPv6) - -## Troubleshooting - -The [sflowtool][] utility can be used to print sFlow packets, and compared -against the metrics produced by Telegraf. - -```sh -sflowtool -p 6343 -``` - -If opening an issue, in addition to the output of sflowtool it will also be -helpful to collect a packet capture. Adjust the interface, host and port as -needed: - -```sh -sudo tcpdump -s 0 -i eth0 -w telegraf-sflow.pcap host 127.0.0.1 and port 6343 -``` - -[sflowtool]: https://github.com/sflow/sflowtool - -## Example Output - -```text -sflow,agent_address=0.0.0.0,dst_ip=10.0.0.2,dst_mac=ff:ff:ff:ff:ff:ff,dst_port=40042,ether_type=IPv4,header_protocol=ETHERNET-ISO88023,input_ifindex=6,ip_dscp=27,ip_ecn=0,output_ifindex=1073741823,source_id_index=3,source_id_type=0,src_ip=10.0.0.1,src_mac=ff:ff:ff:ff:ff:ff,src_port=443 bytes=1570i,drops=0i,frame_length=157i,header_length=128i,ip_flags=2i,ip_fragment_offset=0i,ip_total_length=139i,ip_ttl=42i,sampling_rate=10i,tcp_header_length=0i,tcp_urgent_pointer=0i,tcp_window_size=14i 1584473704793580447 -``` diff --git a/plugins/inputs/sflow/binaryio/minreader.go b/plugins/inputs/sflow/binaryio/minreader.go deleted file mode 100644 index fdf6da439f30f..0000000000000 --- a/plugins/inputs/sflow/binaryio/minreader.go +++ /dev/null @@ -1,37 +0,0 @@ -package binaryio - -import "io" - -// MinimumReader is the implementation for MinReader. -type MinimumReader struct { - reader io.Reader - minNumberOfBytesToRead int64 // Min number of bytes we need to read from the reader -} - -// MinReader reads from the reader but ensures there is at least N bytes read from the reader. -// The reader should call Close() when they are done reading. -// Closing the MinReader will read and discard any unread bytes up to minNumberOfBytesToRead. -// CLosing the MinReader does NOT close the underlying reader. -// The underlying implementation is a MinimumReader, which implements ReaderCloser. -func MinReader(r io.Reader, minNumberOfBytesToRead int64) *MinimumReader { - return &MinimumReader{ - reader: r, - minNumberOfBytesToRead: minNumberOfBytesToRead, - } -} - -func (r *MinimumReader) Read(p []byte) (n int, err error) { - n, err = r.reader.Read(p) - r.minNumberOfBytesToRead -= int64(n) - return n, err -} - -// Close does not close the underlying reader, only the MinimumReader -func (r *MinimumReader) Close() error { - if r.minNumberOfBytesToRead > 0 { - b := make([]byte, r.minNumberOfBytesToRead) - _, err := r.reader.Read(b) - return err - } - return nil -} diff --git a/plugins/inputs/sflow/binaryio/minreader_test.go b/plugins/inputs/sflow/binaryio/minreader_test.go deleted file mode 100644 index 0af78c3855ed9..0000000000000 --- a/plugins/inputs/sflow/binaryio/minreader_test.go +++ /dev/null @@ -1,39 +0,0 @@ -package binaryio - -import ( - "bytes" - "testing" -) - -func TestMinReader(t *testing.T) { - b := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15} - r := bytes.NewBuffer(b) - - mr := MinReader(r, 10) - - toRead := make([]byte, 5) - n, err := mr.Read(toRead) - if err != nil { - t.Error(err) - } - if n != 5 { - t.Error("Expected n to be 5, but was ", n) - } - if !bytes.Equal(toRead, []byte{1, 2, 3, 4, 5}) { - t.Error("expected 5 specific bytes to be read") - } - err = mr.Close() - if err != nil { - t.Error(err) - } - n, err = r.Read(toRead) // read from the outer stream - if err != nil { - t.Error(err) - } - if n != 5 { - t.Error("Expected n to be 5, but was ", n) - } - if !bytes.Equal(toRead, []byte{11, 12, 13, 14, 15}) { - t.Error("expected the last 5 bytes to be read") - } -} diff --git a/plugins/inputs/sflow/decoder_test.go b/plugins/inputs/sflow/decoder_test.go deleted file mode 100644 index 3184fb8b6dc09..0000000000000 --- a/plugins/inputs/sflow/decoder_test.go +++ /dev/null @@ -1,841 +0,0 @@ -package sflow - -import ( - "bytes" - "encoding/hex" - "testing" - "time" - - "github.com/stretchr/testify/require" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/metric" - "github.com/influxdata/telegraf/testutil" -) - -func TestIPv4SW(t *testing.T) { - str := `00000005` + // version - `00000001` + // address type - `c0a80102` + // ip address - `00000010` + // sub agent id - `0000f3d4` + // sequence number - `0bfa047f` + // uptime - `00000002` + // sample count - `00000001` + // sample type - `000000d0` + // sample data length - `0001210a` + // sequence number - `000001fe` + // source id 00 = source id type, 0001fe = source id index - `00000400` + // sampling rate.. apparently this should be input if index???? - `04842400` + // sample pool - `00000000` + // drops - `000001fe` + // input if index - `00000200` + // output if index - `00000002` + // flow records count - `00000001` + // FlowFormat - `00000090` + // flow length - `00000001` + // header protocol - `0000010b` + // Frame length - `00000004` + // stripped octets - `00000080` + // header length - `000c2936d3d6` + // dest mac - `94c691aa9760` + // source mac - `0800` + // etype code: ipv4 - `4500` + // dscp + ecn - `00f9` + // total length - `f190` + // identification - `4000` + // fragment offset + flags - `40` + // ttl - `11` + // protocol - `b4f5` + // header checksum - `c0a80913` + // source ip - `c0a8090a` + // dest ip - `00a1` + // source port - `ba05` + // dest port - `00e5` + // udp length - // rest of header/flowSample we ignore - `641f3081da02010104066d6f746f6770a281cc02047b46462e0201000201003081bd3012060d2b06010201190501010281dc710201003013060d2b060` + - `10201190501010281e66802025acc3012060d2b0601020119050101` + - // next flow record - ignored - `000003e90000001000000009000000000000000900000000` + - // next sample - `00000001000000d00000e3cc000002100000400048eb74000000000000000210000002000000000200000001000000900000000100000097000000040` + - `0000080000c2936d3d6fcecda44008f81000009080045000081186440003f119098c0a80815c0a8090a9a690202006d23083c33303e41707220313120` + - `30393a33333a3031206b6e6f64653120736e6d70645b313039385d3a20436f6e6e656374696f6e2066726f6d205544503a205b3139322e3136382e392` + - `e31305d3a34393233362d000003e90000001000000009000000000000000900000000` - packet, err := hex.DecodeString(str) - require.NoError(t, err) - - actual := make([]telegraf.Metric, 0) - dc := newDecoder() - dc.onPacket(func(p *v5Format) { - metrics := makeMetrics(p) - actual = append(actual, metrics...) - }) - buf := bytes.NewReader(packet) - err = dc.decode(buf) - require.NoError(t, err) - - expected := []telegraf.Metric{ - metric.New( - "sflow", - map[string]string{ - "agent_address": "192.168.1.2", - "dst_ip": "192.168.9.10", - "dst_mac": "00:0c:29:36:d3:d6", - "dst_port": "47621", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "510", - "output_ifindex": "512", - "sample_direction": "ingress", - "source_id_index": "510", - "source_id_type": "0", - "src_ip": "192.168.9.19", - "src_mac": "94:c6:91:aa:97:60", - "src_port": "161", - }, - map[string]interface{}{ - "bytes": uint64(0x042c00), - "drops": uint64(0x00), - "frame_length": uint64(0x010b), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0xf9), - "ip_ttl": uint64(0x40), - "sampling_rate": uint64(0x0400), - "udp_length": uint64(0xe5), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "192.168.1.2", - "dst_ip": "192.168.9.10", - "dst_mac": "00:0c:29:36:d3:d6", - "dst_port": "514", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "528", - "output_ifindex": "512", - "sample_direction": "ingress", - "source_id_index": "528", - "source_id_type": "0", - "src_ip": "192.168.8.21", - "src_mac": "fc:ec:da:44:00:8f", - "src_port": "39529", - }, - map[string]interface{}{ - "bytes": uint64(0x25c000), - "drops": uint64(0x00), - "frame_length": uint64(0x97), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x81), - "ip_ttl": uint64(0x3f), - "sampling_rate": uint64(0x4000), - "udp_length": uint64(0x6d), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} - -func BenchmarkDecodeIPv4SW(b *testing.B) { - packet, err := hex.DecodeString( - "0000000500000001c0a80102000000100000f3d40bfa047f0000000200000001000000d00001210a000001fe000004000484240000000000000001fe0" + - "0000200000000020000000100000090000000010000010b0000000400000080000c2936d3d694c691aa97600800450000f9f19040004011b4f5c0a80" + - "913c0a8090a00a1ba0500e5641f3081da02010104066d6f746f6770a281cc02047b46462e0201000201003081bd3012060d2b0601020119050101028" + - "1dc710201003013060d2b06010201190501010281e66802025acc3012060d2b0601020119050101000003e9000000100000000900000000000000090" + - "000000000000001000000d00000e3cc000002100000400048eb740000000000000002100000020000000002000000010000009000000001000000970" + - "000000400000080000c2936d3d6fcecda44008f81000009080045000081186440003f119098c0a80815c0a8090a9a690202006d23083c33303e41707" + - "22031312030393a33333a3031206b6e6f64653120736e6d70645b313039385d3a20436f6e6e656374696f6e2066726f6d205544503a205b3139322e3" + - "136382e392e31305d3a34393233362d000003e90000001000000009000000000000000900000000", - ) - require.NoError(b, err) - - dc := newDecoder() - require.NoError(b, err) - - b.ResetTimer() - for n := 0; n < b.N; n++ { - _, err = dc.decodeOnePacket(bytes.NewBuffer(packet)) - if err != nil { - panic(err) - } - } -} - -func TestExpandFlow(t *testing.T) { - packet, err := hex.DecodeString( - "00000005000000010a00015000000000000f58998ae119780000000300000003000000c4000b62a90000000000100c840000040024fb7e1e000000000" + - "0000000001017840000000000100c8400000001000000010000009000000001000005bc0000000400000080001b17000130001201f58d44810023710" + - "800450205a6305440007e06ee92ac100016d94d52f505997e701fa1e17aff62574a50100200355f000000ffff00000b004175746f72697a7a6174610" + - "400008040ffff000400008040050031303030320500313030302004000000000868a200000000000000000860a200000000000000000003000000c40" + - "003cecf000000000010170400004000a168ac1c000000000000000000101784000000000010170400000001000000010000009000000001000005f20" + - "0000004000000800024e8324338d4ae52aa0b54810020060800450005dc5420400080061397c0a8060cc0a806080050efcfbb25bad9a21c839a50100" + - "0fff54000008a55f70975a0ff88b05735597ae274bd81fcba17e6e9206b8ea0fb07d05fc27dad06cfe3fdba5d2fc4d057b0add711e596cbe5e9b4bbe" + - "8be59cd77537b7a89f7414a628b736d00000003000000c0000c547a0000000000100c04000004005bc3c3b5000000000000000000101784000000000" + - "0100c0400000001000000010000008c000000010000007e000000040000007a001b17000130001201f58d448100237108004500006824ea4000ff32c" + - "326d94d5105501018f02e88d003000001dd39b1d025d1c68689583b2ab21522d5b5a959642243804f6d51e63323091cc04544285433eb3f6b29e1046" + - "a6a2fa7806319d62041d8fa4bd25b7cd85b8db54202054a077ac11de84acbe37a550004", - ) - require.NoError(t, err) - - dc := newDecoder() - p, err := dc.decodeOnePacket(bytes.NewBuffer(packet)) - require.NoError(t, err) - actual := makeMetrics(p) - - expected := []telegraf.Metric{ - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "217.77.82.245", - "dst_mac": "00:1b:17:00:01:30", - "dst_port": "32368", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1054596", - "output_ifindex": "1051780", - "sample_direction": "egress", - "source_id_index": "1051780", - "source_id_type": "0", - "src_ip": "172.16.0.22", - "src_mac": "00:12:01:f5:8d:44", - "src_port": "1433", - }, - map[string]interface{}{ - "bytes": uint64(0x16f000), - "drops": uint64(0x00), - "frame_length": uint64(0x05bc), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x05a6), - "ip_ttl": uint64(0x7e), - "sampling_rate": uint64(0x0400), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0x0200), - "ip_dscp": "0", - "ip_ecn": "2", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "192.168.6.8", - "dst_mac": "00:24:e8:32:43:38", - "dst_port": "61391", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1054596", - "output_ifindex": "1054468", - "sample_direction": "egress", - "source_id_index": "1054468", - "source_id_type": "0", - "src_ip": "192.168.6.12", - "src_mac": "d4:ae:52:aa:0b:54", - "src_port": "80", - }, - map[string]interface{}{ - "bytes": uint64(0x017c8000), - "drops": uint64(0x00), - "frame_length": uint64(0x05f2), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x05dc), - "ip_ttl": uint64(0x80), - "sampling_rate": uint64(0x4000), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0xff), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "80.16.24.240", - "dst_mac": "00:1b:17:00:01:30", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1054596", - "output_ifindex": "1051652", - "sample_direction": "egress", - "source_id_index": "1051652", - "source_id_type": "0", - "src_ip": "217.77.81.5", - "src_mac": "00:12:01:f5:8d:44", - }, - map[string]interface{}{ - "bytes": uint64(0x01f800), - "drops": uint64(0x00), - "frame_length": uint64(0x7e), - "header_length": uint64(0x7a), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x68), - "ip_ttl": uint64(0xff), - "sampling_rate": uint64(0x0400), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} - -func TestIPv4SWRT(t *testing.T) { - packet, err := hex.DecodeString( - "000000050000000189dd4f010000000000003d4f21151ad40000000600000001000000bc354b97090000020c000013b175792bea000000000000028f0" + - "000020c0000000300000001000000640000000100000058000000040000005408b2587a57624c16fc0b61a5080045000046c3e440003a1118a0052aa" + - "da7569e5ab367a6e35b0032d7bbf1f2fb2eb2490a97f87abc31e135834be367000002590000ffffffffffffffff02add830d51e0aec14cf000003e90" + - "000001000000000000000000000000000000000000003ea0000001000000001c342e32a000000160000000b00000001000000a88b8ffb57000002a20" + - "00013b12e344fd800000000000002a20000028f0000000300000001000000500000000100000042000000040000003e4c16fc0b6202c03e0fdecafe0" + - "80045000030108000007d11fe45575185a718693996f0570e8c001c20614ad602003fd6d4afa6a6d18207324000271169b00000000003e9000000100" + - "0000000000000000000000000000000000003ea000000100000000189dd4f210000000f0000001800000001000000e8354b970a0000020c000013b17" + - "5793f9b000000000000028f0000020c00000003000000010000009000000001000001a500000004000000800231466d0b2c4c16fc0b61a5080045000" + - "193198f40003a114b75052aae1f5f94c778678ef24d017f50ea7622287c30799e1f7d45932d01ca92c46d930000927c0000ffffffffffffffff02ad0" + - "eea6498953d1c7ebb6dbdf0525c80e1a9a62bacfea92f69b7336c2f2f60eba0593509e14eef167eb37449f05ad70b8241c1a46d000003e9000000100" + - "0000000000000000000000000000000000003ea0000001000000001c342e1fd000000160000001000000001000000e8354b970b0000020c000013b17" + - "579534c000000000000028f0000020c00000003000000010000009000000001000000b500000004000000800231466d0b2c4c16fc0b61a5080045000" + - "0a327c240003606fd67b93c706a021ff365045fe8a0976d624df8207083501800edb31b0000485454502f312e3120323030204f4b0d0a53657276657" + - "23a2050726f746f636f6c20485454500d0a436f6e74656e742d4c656e6774683a20313430340d0a436f6e6e656374696f6e3a20000003e9000000100" + - "0000000000000000000000000000000000003ea0000001000000001c342e1fd000000170000001000000001000000e8354b970c0000020c000013b17" + - "57966fd000000000000028f0000020c000000030000000100000090000000010000018e00000004000000800231466d0b2c4c16fc0b61a5080045000" + - "17c7d2c40003a116963052abd8d021c940e67e7e0d501682342dbe7936bd47ef487dee5591ec1b24d83622e000072250000ffffffffffffffff02ad0" + - "039d8ba86a90017071d76b177de4d8c4e23bcaaaf4d795f77b032f959e0fb70234d4c28922d4e08dd3330c66e34bff51cc8ade5000003e9000000100" + - "0000000000000000000000000000000000003ea0000001000000001c342e1fd000000160000001000000001000000e80d6146ac000002a1000013b17" + - "880b49d00000000000002a10000028f00000003000000010000009000000001000005ee00000004000000804c16fc0b6201d8b122766a2c080045000" + - "5dc04574000770623a11fcd80a218691d4cf2fe01bbd4f47482065fd63a5010fabd7987000052a20002c8c43ea91ca1eaa115663f5218a37fbb409df" + - "bbedff54731ef41199b35535905ac2366a05a803146ced544abf45597f3714327d59f99e30c899c39fc5a4b67d12087bf8db2bc000003e9000000100" + - "0000000000000000000000000000000000003ea000000100000000189dd4f210000001000000018", - ) - require.NoError(t, err) - - dc := newDecoder() - p, err := dc.decodeOnePacket(bytes.NewBuffer(packet)) - require.NoError(t, err) - actual := makeMetrics(p) - - expected := []telegraf.Metric{ - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "86.158.90.179", - "dst_mac": "08:b2:58:7a:57:62", - "dst_port": "58203", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "655", - "output_ifindex": "524", - "sample_direction": "egress", - "source_id_index": "524", - "source_id_type": "0", - "src_ip": "5.42.173.167", - "src_mac": "4c:16:fc:0b:61:a5", - "src_port": "26534", - }, - map[string]interface{}{ - "bytes": uint64(0x06c4d8), - "drops": uint64(0x00), - "frame_length": uint64(0x58), - "header_length": uint64(0x54), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x46), - "ip_ttl": uint64(0x3a), - "sampling_rate": uint64(0x13b1), - "udp_length": uint64(0x32), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "24.105.57.150", - "dst_mac": "4c:16:fc:0b:62:02", - "dst_port": "3724", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "674", - "output_ifindex": "655", - "sample_direction": "ingress", - "source_id_index": "674", - "source_id_type": "0", - "src_ip": "87.81.133.167", - "src_mac": "c0:3e:0f:de:ca:fe", - "src_port": "61527", - }, - map[string]interface{}{ - "bytes": uint64(0x0513a2), - "drops": uint64(0x00), - "frame_length": uint64(0x42), - "header_length": uint64(0x3e), - "ip_flags": uint64(0x00), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x30), - "ip_ttl": uint64(0x7d), - "sampling_rate": uint64(0x13b1), - "udp_length": uint64(0x1c), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "95.148.199.120", - "dst_mac": "02:31:46:6d:0b:2c", - "dst_port": "62029", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "655", - "output_ifindex": "524", - "sample_direction": "egress", - "source_id_index": "524", - "source_id_type": "0", - "src_ip": "5.42.174.31", - "src_mac": "4c:16:fc:0b:61:a5", - "src_port": "26510", - }, - map[string]interface{}{ - "bytes": uint64(0x206215), - "drops": uint64(0x00), - "frame_length": uint64(0x01a5), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x0193), - "ip_ttl": uint64(0x3a), - "sampling_rate": uint64(0x13b1), - "udp_length": uint64(0x017f), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "2.31.243.101", - "dst_mac": "02:31:46:6d:0b:2c", - "dst_port": "59552", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "655", - "output_ifindex": "524", - "sample_direction": "egress", - "source_id_index": "524", - "source_id_type": "0", - "src_ip": "185.60.112.106", - "src_mac": "4c:16:fc:0b:61:a5", - "src_port": "1119", - }, - map[string]interface{}{ - "bytes": uint64(0x0dec25), - "drops": uint64(0x00), - "frame_length": uint64(0xb5), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0xa3), - "ip_ttl": uint64(0x36), - "sampling_rate": uint64(0x13b1), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0xed), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "2.28.148.14", - "dst_mac": "02:31:46:6d:0b:2c", - "dst_port": "57557", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "655", - "output_ifindex": "524", - "sample_direction": "egress", - "source_id_index": "524", - "source_id_type": "0", - "src_ip": "5.42.189.141", - "src_mac": "4c:16:fc:0b:61:a5", - "src_port": "26599", - }, - map[string]interface{}{ - "bytes": uint64(0x1e9d2e), - "drops": uint64(0x00), - "frame_length": uint64(0x018e), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x017c), - "ip_ttl": uint64(0x3a), - "sampling_rate": uint64(0x13b1), - "udp_length": uint64(0x0168), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "137.221.79.1", - "dst_ip": "24.105.29.76", - "dst_mac": "4c:16:fc:0b:62:01", - "dst_port": "443", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "673", - "output_ifindex": "655", - "sample_direction": "ingress", - "source_id_index": "673", - "source_id_type": "0", - "src_ip": "31.205.128.162", - "src_mac": "d8:b1:22:76:6a:2c", - "src_port": "62206", - }, - map[string]interface{}{ - "bytes": uint64(0x74c38e), - "drops": uint64(0x00), - "frame_length": uint64(0x05ee), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x05dc), - "ip_ttl": uint64(0x77), - "sampling_rate": uint64(0x13b1), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0xfabd), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} - -func TestIPv6SW(t *testing.T) { - packet, err := hex.DecodeString( - "00000005000000010ae0648100000002000093d824ac82340000000100000001000000d000019f94000001010000100019f9400000000000000001010" + - "0000000000000020000000100000090000000010000058c00000008000000800008e3fffc10d4f4be04612486dd60000000054e113a2607f8b040020" + - "0140000000000000008262000edc000e804a25e30c581af36fa01bbfa6f054e249810b584bcbf12926c2e29a779c26c72db483e8191524fe2288bfda" + - "ceaf9d2e724d04305706efcfdef70db86873bbacf29698affe4e7d6faa21d302f9b4b023291a05a000003e90000001000000001000000000000000100000000", - ) - require.NoError(t, err) - - dc := newDecoder() - p, err := dc.decodeOnePacket(bytes.NewBuffer(packet)) - require.NoError(t, err) - actual := makeMetrics(p) - - expected := []telegraf.Metric{ - - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.224.100.129", - "dst_ip": "2620:ed:c000:e804:a25e:30c5:81af:36fa", - "dst_mac": "00:08:e3:ff:fc:10", - "dst_port": "64111", - "ether_type": "IPv6", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "257", - "output_ifindex": "0", - "sample_direction": "ingress", - "source_id_index": "257", - "source_id_type": "0", - "src_ip": "2607:f8b0:4002:14::8", - "src_mac": "d4:f4:be:04:61:24", - "src_port": "443", - }, - map[string]interface{}{ - "bytes": uint64(0x58c000), - "drops": uint64(0x00), - "frame_length": uint64(0x058c), - "header_length": uint64(0x80), - "sampling_rate": uint64(0x1000), - "payload_length": uint64(0x054e), - "udp_length": uint64(0x054e), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} - -func TestExpandFlowCounter(t *testing.T) { - packet, err := hex.DecodeString( - "00000005000000010a00015000000000000f58898ae0fa380000000700000004000000ec00006ece00000000001017840000000300000002000000340" + - "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000580" + - "01017840000000600000002540be400000000010000000300007b8ebd37b97e61ff94860803e8e908ffb2b500000000000000000000000000018e7c3" + - "1ee7ba4195f041874579ff021ba936300000000000000000000000100000007000000380011223344550003f8b15645e7e7d6960000002fe2fc02fc0" + - "1edbf580000000000000000000000000000000001dcb9cf000000000000000000000004000000ec00006ece000000000010018400000003000000020" + - "000003400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010" + - "0000058001001840000000600000002540be400000000010000000300000841131d1fd9f850bfb103617cb401e659890000000000000000000000000" + - "0000bec1902e5da9212e3e96d7996e922513250000000000000000000000001000000070000003800112233445500005c260acbddb3000100000003e" + - "2fc02fc01ee414f0000000000000000000000000000000001dccdd30000000000000000000000030000008400004606000000000010030400004000a" + - "d9dc19b0000000000000000001017840000000000100304000000010000000100000050000000010000004400000004000000400012815116c400151" + - "7cf426d8100200608004500002895da40008006d74bc0a8060ac0a8064f04ef04aab1797122cf7eaf4f5010ffff77270000000000000000000000030" + - "00000b0001bd698000000000010148400000400700b180f000000000000000000101504000000000010148400000001000000010000007c000000010" + - "000006f000000040000006b001b17000131f0f755b9afc081000439080045000059045340005206920c1f0d4703d94d52e201bbf14977d1e9f15498a" + - "f36801800417f1100000101080afdf3c70400e043871503010020ff268cfe2e2fd5fffe1d3d704a91d57b895f174c4b4428c66679d80a307294303f0" + - "0000003000000c40003ceca000000000010170400004000a166aa7a00000000000000000010178400000000001017040000000100000001000000900" + - "0000001000005f200000004000000800024e8369e2bd4ae52aa0b54810020060800450005dc4c71400080061b45c0a8060cc0a806090050f855692a7" + - "a94a1154ae1801001046b6a00000101080a6869a48d151016d046a84a7aa1c6743fa05179f7ecbd4e567150cb6f2077ff89480ae730637d26d2237c0" + - "8548806f672c7476eb1b5a447b42cb9ce405994d152fa3e000000030000008c001bd699000000000010148400000400700b180f00000000000000000" + - "01015040000000000101484000000010000000100000058000000010000004a0000000400000046001b17000131f0f755b9afc081000439080045000" + - "0340ce040003a06bea5c1ce8793d94d528f00504c3b08b18f275b83d5df8010054586ad00000101050a5b83d5de5b83d5df11d800000003000000c40" + - "0004e07000000000010028400004000c7ec97f2000000000000000000100784000000000010028400000001000000010000009000000001000005f20" + - "00000040000008000005e0001ff005056800dd18100000a0800450005dc5a42400040066ef70a000ac8c0a8967201bbe17c81597908caf8a05f50100" + - "10328610000f172263da0ba5d6223c079b8238bc841256bf17c4ffb08ad11c4fbff6f87ae1624a6b057b8baa9342114e5f5b46179083020cb560c4e9" + - "eadcec6dfd83e102ddbc27024803eb5", - ) - require.NoError(t, err) - - dc := newDecoder() - p, err := dc.decodeOnePacket(bytes.NewBuffer(packet)) - require.NoError(t, err) - actual := makeMetrics(p) - - expected := []telegraf.Metric{ - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "192.168.6.79", - "dst_mac": "00:12:81:51:16:c4", - "dst_port": "1194", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1054596", - "output_ifindex": "1049348", - "sample_direction": "egress", - "source_id_index": "1049348", - "source_id_type": "0", - "src_ip": "192.168.6.10", - "src_mac": "00:15:17:cf:42:6d", - "src_port": "1263", - }, - map[string]interface{}{ - "bytes": uint64(0x110000), - "drops": uint64(0x00), - "frame_length": uint64(0x44), - "header_length": uint64(0x40), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x28), - "ip_ttl": uint64(0x80), - "sampling_rate": uint64(0x4000), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0xffff), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "217.77.82.226", - "dst_mac": "00:1b:17:00:01:31", - "dst_port": "61769", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1053956", - "output_ifindex": "1053828", - "sample_direction": "egress", - "source_id_index": "1053828", - "source_id_type": "0", - "src_ip": "31.13.71.3", - "src_mac": "f0:f7:55:b9:af:c0", - "src_port": "443", - }, - map[string]interface{}{ - "bytes": uint64(0x01bc00), - "drops": uint64(0x00), - "frame_length": uint64(0x6f), - "header_length": uint64(0x6b), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x59), - "ip_ttl": uint64(0x52), - "sampling_rate": uint64(0x0400), - "tcp_header_length": uint64(0x20), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0x41), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "192.168.6.9", - "dst_mac": "00:24:e8:36:9e:2b", - "dst_port": "63573", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1054596", - "output_ifindex": "1054468", - "sample_direction": "egress", - "source_id_index": "1054468", - "source_id_type": "0", - "src_ip": "192.168.6.12", - "src_mac": "d4:ae:52:aa:0b:54", - "src_port": "80", - }, - map[string]interface{}{ - "bytes": uint64(0x017c8000), - "drops": uint64(0x00), - "frame_length": uint64(0x05f2), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x05dc), - "ip_ttl": uint64(0x80), - "sampling_rate": uint64(0x4000), - "tcp_header_length": uint64(0x20), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0x0104), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "217.77.82.143", - "dst_mac": "00:1b:17:00:01:31", - "dst_port": "19515", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1053956", - "output_ifindex": "1053828", - "sample_direction": "egress", - "source_id_index": "1053828", - "source_id_type": "0", - "src_ip": "193.206.135.147", - "src_mac": "f0:f7:55:b9:af:c0", - "src_port": "80", - }, - map[string]interface{}{ - "bytes": uint64(0x012800), - "drops": uint64(0x00), - "frame_length": uint64(0x4a), - "header_length": uint64(0x46), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x34), - "ip_ttl": uint64(0x3a), - "sampling_rate": uint64(0x0400), - "tcp_header_length": uint64(0x20), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0x0545), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "10.0.1.80", - "dst_ip": "192.168.150.114", - "dst_mac": "00:00:5e:00:01:ff", - "dst_port": "57724", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "1050500", - "output_ifindex": "1049220", - "sample_direction": "egress", - "source_id_index": "1049220", - "source_id_type": "0", - "src_ip": "10.0.10.200", - "src_mac": "00:50:56:80:0d:d1", - "src_port": "443", - }, - map[string]interface{}{ - "bytes": uint64(0x017c8000), - "drops": uint64(0x00), - "frame_length": uint64(0x05f2), - "header_length": uint64(0x80), - "ip_flags": uint64(0x02), - "ip_fragment_offset": uint64(0x00), - "ip_total_length": uint64(0x05dc), - "ip_ttl": uint64(0x40), - "sampling_rate": uint64(0x4000), - "tcp_header_length": uint64(0x14), - "tcp_urgent_pointer": uint64(0x00), - "tcp_window_size": uint64(0x0103), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} - -func TestFlowExpandCounter(t *testing.T) { - packet, err := hex.DecodeString( - "00000005000000010a000150000000000006d14d8ae0fe200000000200000004000000ac00006d15000000004b00ca000000000200000002000000340" + - "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000584" + - "b00ca0000000001000000000000000000000001000000010000308ae33bb950eb92a8a3004d0bb406899571000000000000000000000000000012f7e" + - "d9c9db8c24ed90604eaf0bd04636edb00000000000000000000000100000004000000ac00006d15000000004b0054000000000200000002000000340" + - "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000584" + - "b00540000000001000000003b9aca000000000100000003000067ba8e64fd23fa65f26d0215ec4a0021086600000000000000000000000000002002c" + - "3b21045c2378ad3001fb2f300061872000000000000000000000001", - ) - require.NoError(t, err) - - dc := newDecoder() - p, err := dc.decodeOnePacket(bytes.NewBuffer(packet)) - require.NoError(t, err) - actual := makeMetrics(p) - - // we don't do anything with samples yet - expected := make([]telegraf.Metric, 0) - testutil.RequireMetricsEqual(t, expected, actual, testutil.IgnoreTime()) -} diff --git a/plugins/inputs/sflow/metricencoder.go b/plugins/inputs/sflow/metricencoder.go deleted file mode 100644 index 0a85418b2932d..0000000000000 --- a/plugins/inputs/sflow/metricencoder.go +++ /dev/null @@ -1,43 +0,0 @@ -package sflow - -import ( - "strconv" - "time" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/metric" -) - -func makeMetrics(p *v5Format) []telegraf.Metric { - now := time.Now() - metrics := make([]telegraf.Metric, 0) - tags := map[string]string{ - "agent_address": p.agentAddress.String(), - } - fields := make(map[string]interface{}, 2) - for _, sample := range p.samples { - tags["input_ifindex"] = strconv.FormatUint(uint64(sample.smplData.inputIfIndex), 10) - tags["output_ifindex"] = strconv.FormatUint(uint64(sample.smplData.outputIfIndex), 10) - tags["sample_direction"] = sample.smplData.sampleDirection - tags["source_id_index"] = strconv.FormatUint(uint64(sample.smplData.sourceIDIndex), 10) - tags["source_id_type"] = strconv.FormatUint(uint64(sample.smplData.sourceIDType), 10) - fields["drops"] = sample.smplData.drops - fields["sampling_rate"] = sample.smplData.samplingRate - - for _, flowRecord := range sample.smplData.flowRecords { - if flowRecord.flowData != nil { - tags2 := flowRecord.flowData.getTags() - fields2 := flowRecord.flowData.getFields() - for k, v := range tags { - tags2[k] = v - } - for k, v := range fields { - fields2[k] = v - } - m := metric.New("sflow", tags2, fields2, now) - metrics = append(metrics, m) - } - } - } - return metrics -} diff --git a/plugins/inputs/sflow/packetdecoder.go b/plugins/inputs/sflow/packetdecoder.go deleted file mode 100644 index cf6cfe92354e7..0000000000000 --- a/plugins/inputs/sflow/packetdecoder.go +++ /dev/null @@ -1,480 +0,0 @@ -package sflow - -import ( - "encoding/binary" - "errors" - "fmt" - "io" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/plugins/inputs/sflow/binaryio" -) - -type packetDecoder struct { - onPacketF func(p *v5Format) - Log telegraf.Logger -} - -func newDecoder() *packetDecoder { - return &packetDecoder{} -} - -func (d *packetDecoder) debug(args ...interface{}) { - if d.Log != nil { - d.Log.Debug(args...) - } -} - -func (d *packetDecoder) onPacket(f func(p *v5Format)) { - d.onPacketF = f -} - -func (d *packetDecoder) decode(r io.Reader) error { - var err error - var packet *v5Format - for err == nil { - packet, err = d.decodeOnePacket(r) - if err != nil { - break - } - d.onPacketF(packet) - } - if err != nil && errors.Is(err, io.EOF) { - return nil - } - return err -} - -type addressType uint32 // must be uint32 - -const ( - addressTypeUnknown addressType = 0 - addressTypeIPV4 addressType = 1 - addressTypeIPV6 addressType = 2 -) - -func (d *packetDecoder) decodeOnePacket(r io.Reader) (*v5Format, error) { - p := &v5Format{} - err := read(r, &p.version, "version") - if err != nil { - return nil, err - } - if p.version != 5 { - return nil, fmt.Errorf("version %d not supported, only version 5", p.version) - } - var addressIPType addressType - if err := read(r, &addressIPType, "address ip type"); err != nil { - return nil, err - } - switch addressIPType { - case addressTypeUnknown: - p.agentAddress.IP = make([]byte, 0) - case addressTypeIPV4: - p.agentAddress.IP = make([]byte, 4) - case addressTypeIPV6: - p.agentAddress.IP = make([]byte, 16) - default: - return nil, fmt.Errorf("unknown address IP type %d", addressIPType) - } - if err := read(r, &p.agentAddress.IP, "Agent Address IP"); err != nil { - return nil, err - } - if err := read(r, &p.subAgentID, "SubAgentID"); err != nil { - return nil, err - } - if err := read(r, &p.sequenceNumber, "SequenceNumber"); err != nil { - return nil, err - } - if err := read(r, &p.uptime, "Uptime"); err != nil { - return nil, err - } - - p.samples, err = d.decodeSamples(r) - return p, err -} - -func (d *packetDecoder) decodeSamples(r io.Reader) ([]sample, error) { - // # of samples - var numOfSamples uint32 - if err := read(r, &numOfSamples, "sample count"); err != nil { - return nil, err - } - - result := make([]sample, 0, numOfSamples) - for i := 0; i < int(numOfSamples); i++ { - sam, err := d.decodeSample(r) - if err != nil { - return result, err - } - result = append(result, sam) - } - - return result, nil -} - -func (d *packetDecoder) decodeSample(r io.Reader) (sample, error) { - var err error - sam := sample{} - if err := read(r, &sam.smplType, "sampleType"); err != nil { - return sam, err - } - sampleDataLen := uint32(0) - if err := read(r, &sampleDataLen, "Sample data length"); err != nil { - return sam, err - } - mr := binaryio.MinReader(r, int64(sampleDataLen)) - defer mr.Close() - - switch sam.smplType { - case sampleTypeFlowSample: - sam.smplData, err = d.decodeFlowSample(mr) - case sampleTypeFlowSampleExpanded: - sam.smplData, err = d.decodeFlowSampleExpanded(mr) - default: - d.debug("Unknown sample type: ", sam.smplType) - } - return sam, err -} - -func (d *packetDecoder) decodeFlowSample(r io.Reader) (t sampleDataFlowSampleExpanded, err error) { - if err := read(r, &t.sequenceNumber, "SequenceNumber"); err != nil { - return t, err - } - var sourceID uint32 - if err := read(r, &sourceID, "SourceID"); err != nil { // source_id sflow_version_5.txt line: 1622 - return t, err - } - // split source id to source id type and source id index - t.sourceIDIndex = sourceID & 0x00ffffff // sflow_version_5.txt line: 1468 - t.sourceIDType = sourceID >> 24 // source_id_type sflow_version_5.txt Line 1465 - if err := read(r, &t.samplingRate, "SamplingRate"); err != nil { - return t, err - } - if err := read(r, &t.samplePool, "SamplePool"); err != nil { - return t, err - } - if err := read(r, &t.drops, "Drops"); err != nil { // sflow_version_5.txt line 1636 - return t, err - } - if err := read(r, &t.inputIfIndex, "InputIfIndex"); err != nil { - return t, err - } - t.inputIfFormat = t.inputIfIndex >> 30 - t.inputIfIndex = t.inputIfIndex & 0x3FFFFFFF - - if err := read(r, &t.outputIfIndex, "OutputIfIndex"); err != nil { - return t, err - } - t.outputIfFormat = t.outputIfIndex >> 30 - t.outputIfIndex = t.outputIfIndex & 0x3FFFFFFF - - switch t.sourceIDIndex { - case t.outputIfIndex: - t.sampleDirection = "egress" - case t.inputIfIndex: - t.sampleDirection = "ingress" - } - - t.flowRecords, err = d.decodeFlowRecords(r, t.samplingRate) - return t, err -} - -func (d *packetDecoder) decodeFlowSampleExpanded(r io.Reader) (t sampleDataFlowSampleExpanded, err error) { - if err := read(r, &t.sequenceNumber, "SequenceNumber"); err != nil { // sflow_version_5.txt line 1701 - return t, err - } - if err := read(r, &t.sourceIDType, "SourceIDType"); err != nil { // sflow_version_5.txt line: 1706 + 16878 - return t, err - } - if err := read(r, &t.sourceIDIndex, "SourceIDIndex"); err != nil { // sflow_version_5.txt line: 1689 - return t, err - } - if err := read(r, &t.samplingRate, "SamplingRate"); err != nil { // sflow_version_5.txt line: 1707 - return t, err - } - if err := read(r, &t.samplePool, "SamplePool"); err != nil { // sflow_version_5.txt line: 1708 - return t, err - } - if err := read(r, &t.drops, "Drops"); err != nil { // sflow_version_5.txt line: 1712 - return t, err - } - if err := read(r, &t.inputIfFormat, "InputIfFormat"); err != nil { // sflow_version_5.txt line: 1727 - return t, err - } - if err := read(r, &t.inputIfIndex, "InputIfIndex"); err != nil { - return t, err - } - if err := read(r, &t.outputIfFormat, "OutputIfFormat"); err != nil { // sflow_version_5.txt line: 1728 - return t, err - } - if err := read(r, &t.outputIfIndex, "OutputIfIndex"); err != nil { - return t, err - } - - switch t.sourceIDIndex { - case t.outputIfIndex: - t.sampleDirection = "egress" - case t.inputIfIndex: - t.sampleDirection = "ingress" - } - - t.flowRecords, err = d.decodeFlowRecords(r, t.samplingRate) - return t, err -} - -func (d *packetDecoder) decodeFlowRecords(r io.Reader, samplingRate uint32) (recs []flowRecord, err error) { - var flowDataLen uint32 - var count uint32 - if err := read(r, &count, "FlowRecord count"); err != nil { - return recs, err - } - for i := uint32(0); i < count; i++ { - fr := flowRecord{} - if err := read(r, &fr.flowFormat, "FlowFormat"); err != nil { // sflow_version_5.txt line 1597 - return recs, err - } - if err := read(r, &flowDataLen, "Flow data length"); err != nil { - return recs, err - } - - mr := binaryio.MinReader(r, int64(flowDataLen)) - - switch fr.flowFormat { - case flowFormatTypeRawPacketHeader: // sflow_version_5.txt line 1938 - fr.flowData, err = d.decodeRawPacketHeaderFlowData(mr, samplingRate) - default: - d.debug("Unknown flow format: ", fr.flowFormat) - } - if err != nil { - mr.Close() - return recs, err - } - - recs = append(recs, fr) - mr.Close() - } - - return recs, err -} - -func (d *packetDecoder) decodeRawPacketHeaderFlowData(r io.Reader, samplingRate uint32) (h rawPacketHeaderFlowData, err error) { - if err := read(r, &h.headerProtocol, "HeaderProtocol"); err != nil { // sflow_version_5.txt line 1940 - return h, err - } - if err := read(r, &h.frameLength, "FrameLength"); err != nil { // sflow_version_5.txt line 1942 - return h, err - } - h.bytes = h.frameLength * samplingRate - - if err := read(r, &h.strippedOctets, "StrippedOctets"); err != nil { // sflow_version_5.txt line 1967 - return h, err - } - if err := read(r, &h.headerLength, "HeaderLength"); err != nil { - return h, err - } - - mr := binaryio.MinReader(r, int64(h.headerLength)) - defer mr.Close() - - switch h.headerProtocol { - case headerProtocolTypeEthernetISO88023: - h.header, err = d.decodeEthHeader(mr) - default: - d.debug("Unknown header protocol type: ", h.headerProtocol) - } - - return h, err -} - -// ethHeader answers a decode Directive that will decode an ethernet frame header -// according to https://en.wikipedia.org/wiki/Ethernet_frame -func (d *packetDecoder) decodeEthHeader(r io.Reader) (h ethHeader, err error) { - // we may have to read out StrippedOctets bytes and throw them away first? - if err := read(r, &h.destinationMAC, "DestinationMAC"); err != nil { - return h, err - } - if err := read(r, &h.sourceMAC, "SourceMAC"); err != nil { - return h, err - } - var tagOrEType uint16 - if err := read(r, &tagOrEType, "tagOrEtype"); err != nil { - return h, err - } - switch tagOrEType { - case 0x8100: // could be? - var discard uint16 - if err := read(r, &discard, "unknown"); err != nil { - return h, err - } - if err := read(r, &h.etherTypeCode, "EtherTypeCode"); err != nil { - return h, err - } - default: - h.etherTypeCode = tagOrEType - } - h.etherType = eTypeMap[h.etherTypeCode] - switch h.etherType { - case "IPv4": - h.ipHeader, err = d.decodeIPv4Header(r) - case "IPv6": - h.ipHeader, err = d.decodeIPv6Header(r) - default: - } - if err != nil { - return h, err - } - return h, err -} - -// https://en.wikipedia.org/wiki/IPv4#Header -func (d *packetDecoder) decodeIPv4Header(r io.Reader) (h ipV4Header, err error) { - if err := read(r, &h.version, "Version"); err != nil { - return h, err - } - h.internetHeaderLength = h.version & 0x0F - h.version = h.version & 0xF0 - if err := read(r, &h.dscp, "DSCP"); err != nil { - return h, err - } - h.ecn = h.dscp & 0x03 - h.dscp = h.dscp >> 2 - if err := read(r, &h.totalLength, "TotalLength"); err != nil { - return h, err - } - if err := read(r, &h.identification, "Identification"); err != nil { - return h, err - } - if err := read(r, &h.fragmentOffset, "FragmentOffset"); err != nil { - return h, err - } - h.flags = uint8(h.fragmentOffset >> 13) - h.fragmentOffset = h.fragmentOffset & 0x1FFF - if err := read(r, &h.ttl, "TTL"); err != nil { - return h, err - } - if err := read(r, &h.protocol, "Protocol"); err != nil { - return h, err - } - if err := read(r, &h.headerChecksum, "HeaderChecksum"); err != nil { - return h, err - } - if err := read(r, &h.sourceIP, "SourceIP"); err != nil { - return h, err - } - if err := read(r, &h.destIP, "DestIP"); err != nil { - return h, err - } - switch h.protocol { - case ipProtocolTCP: - h.protocolHeader, err = decodeTCPHeader(r) - case ipProtocolUDP: - h.protocolHeader, err = decodeUDPHeader(r) - default: - d.debug("Unknown IP protocol: ", h.protocol) - } - return h, err -} - -// https://en.wikipedia.org/wiki/IPv6_packet -func (d *packetDecoder) decodeIPv6Header(r io.Reader) (h ipV6Header, err error) { - var fourByteBlock uint32 - if err := read(r, &fourByteBlock, "IPv6 header octet 0"); err != nil { - return h, err - } - version := fourByteBlock >> 28 - if version != 0x6 { - return h, fmt.Errorf("unexpected IPv6 header version 0x%x", version) - } - h.dscp = uint8((fourByteBlock & 0xFC00000) >> 22) - h.ecn = uint8((fourByteBlock & 0x300000) >> 20) - - // The flowLabel is available via fourByteBlock & 0xFFFFF - if err := read(r, &h.payloadLength, "PayloadLength"); err != nil { - return h, err - } - if err := read(r, &h.nextHeaderProto, "NextHeaderProto"); err != nil { - return h, err - } - if err := read(r, &h.hopLimit, "HopLimit"); err != nil { - return h, err - } - if err := read(r, &h.sourceIP, "SourceIP"); err != nil { - return h, err - } - if err := read(r, &h.destIP, "DestIP"); err != nil { - return h, err - } - switch h.nextHeaderProto { - case ipProtocolTCP: - h.protocolHeader, err = decodeTCPHeader(r) - case ipProtocolUDP: - h.protocolHeader, err = decodeUDPHeader(r) - default: - // not handled - d.debug("Unknown IP protocol: ", h.nextHeaderProto) - } - return h, err -} - -// https://en.wikipedia.org/wiki/Transmission_Control_Protocol#TCP_segment_structure -func decodeTCPHeader(r io.Reader) (h tcpHeader, err error) { - if err := read(r, &h.sourcePort, "SourcePort"); err != nil { - return h, err - } - if err := read(r, &h.destinationPort, "DestinationPort"); err != nil { - return h, err - } - if err := read(r, &h.sequence, "Sequence"); err != nil { - return h, err - } - if err := read(r, &h.ackNumber, "AckNumber"); err != nil { - return h, err - } - // Next up: bit reading! - // data offset 4 bits - // reserved 3 bits - // flags 9 bits - var dataOffsetAndReservedAndFlags uint16 - if err := read(r, &dataOffsetAndReservedAndFlags, "TCP Header Octet offset 12"); err != nil { - return h, err - } - h.tcpHeaderLength = uint8((dataOffsetAndReservedAndFlags >> 12) * 4) - h.flags = dataOffsetAndReservedAndFlags & 0x1FF - // done bit reading - - if err := read(r, &h.tcpWindowSize, "TCPWindowSize"); err != nil { - return h, err - } - if err := read(r, &h.checksum, "Checksum"); err != nil { - return h, err - } - if err := read(r, &h.tcpUrgentPointer, "TCPUrgentPointer"); err != nil { - return h, err - } - - return h, err -} - -func decodeUDPHeader(r io.Reader) (h udpHeader, err error) { - if err := read(r, &h.sourcePort, "SourcePort"); err != nil { - return h, err - } - if err := read(r, &h.destinationPort, "DestinationPort"); err != nil { - return h, err - } - if err := read(r, &h.udpLength, "UDPLength"); err != nil { - return h, err - } - if err := read(r, &h.checksum, "Checksum"); err != nil { - return h, err - } - return h, err -} - -func read(r io.Reader, data interface{}, name string) error { - err := binary.Read(r, binary.BigEndian, data) - if err != nil { - return fmt.Errorf("failed to read %q: %w", name, err) - } - return nil -} diff --git a/plugins/inputs/sflow/packetdecoder_test.go b/plugins/inputs/sflow/packetdecoder_test.go deleted file mode 100644 index 41e9f439bc6a7..0000000000000 --- a/plugins/inputs/sflow/packetdecoder_test.go +++ /dev/null @@ -1,205 +0,0 @@ -package sflow - -import ( - "bytes" - "testing" - - "github.com/stretchr/testify/require" -) - -func TestUDPHeader(t *testing.T) { - octets := bytes.NewBuffer([]byte{ - 0x00, 0x01, // src_port - 0x00, 0x02, // dst_port - 0x00, 0x03, // udp_length - 0x00, 0x00, // checksum - }) - - actual, err := decodeUDPHeader(octets) - require.NoError(t, err) - - expected := udpHeader{ - sourcePort: 1, - destinationPort: 2, - udpLength: 3, - } - - require.Equal(t, expected, actual) -} - -func BenchmarkUDPHeader(b *testing.B) { - octets := bytes.NewBuffer([]byte{ - 0x00, 0x01, // src_port - 0x00, 0x02, // dst_port - 0x00, 0x03, // udp_length - 0x00, 0x00, // checksum - }) - - b.ResetTimer() - for n := 0; n < b.N; n++ { - _, err := decodeUDPHeader(octets) - require.NoError(b, err) - } -} - -func TestIPv4Header(t *testing.T) { - octets := bytes.NewBuffer( - []byte{ - 0x45, // version + IHL - 0x00, // ip_dscp + ip_ecn - 0x00, 0x00, // total length - 0x00, 0x00, // identification - 0x00, 0x00, // flags + frag offset - 0x00, // ttl - 0x11, // protocol udp (0x11) - 0x00, 0x00, // header checksum - 0x7f, 0x00, 0x00, 0x01, // src ip - 0x7f, 0x00, 0x00, 0x02, // dst ip - 0x00, 0x01, // src_port - 0x00, 0x02, // dst_port - 0x00, 0x03, // udp_length - 0x00, 0x00, // checksum - }, - ) - dc := newDecoder() - actual, err := dc.decodeIPv4Header(octets) - require.NoError(t, err) - - expected := ipV4Header{ - version: 0x40, - internetHeaderLength: 0x05, - dscp: 0, - ecn: 0, - totalLength: 0, - identification: 0, - flags: 0, - fragmentOffset: 0, - ttl: 0, - protocol: 0x11, - headerChecksum: 0, - sourceIP: [4]byte{127, 0, 0, 1}, - destIP: [4]byte{127, 0, 0, 2}, - protocolHeader: udpHeader{ - sourcePort: 1, - destinationPort: 2, - udpLength: 3, - checksum: 0, - }, - } - - require.Equal(t, expected, actual) -} - -// Using the same Directive instance, prior paths through the parse tree should -// not affect the latest parse. -func TestIPv4HeaderSwitch(t *testing.T) { - octets := bytes.NewBuffer( - []byte{ - 0x45, // version + IHL - 0x00, // ip_dscp + ip_ecn - 0x00, 0x00, // total length - 0x00, 0x00, // identification - 0x00, 0x00, // flags + frag offset - 0x00, // ttl - 0x11, // protocol udp (0x11) - 0x00, 0x00, // header checksum - 0x7f, 0x00, 0x00, 0x01, // src ip - 0x7f, 0x00, 0x00, 0x02, // dst ip - 0x00, 0x01, // src_port - 0x00, 0x02, // dst_port - 0x00, 0x03, // udp_length - 0x00, 0x00, // checksum - }, - ) - dc := newDecoder() - _, err := dc.decodeIPv4Header(octets) - require.NoError(t, err) - - octets = bytes.NewBuffer( - []byte{ - 0x45, // version + IHL - 0x00, // ip_dscp + ip_ecn - 0x00, 0x00, // total length - 0x00, 0x00, // identification - 0x00, 0x00, // flags + frag offset - 0x00, // ttl - 0x06, // protocol tcp (0x06) - 0x00, 0x00, // header checksum - 0x7f, 0x00, 0x00, 0x01, // src ip - 0x7f, 0x00, 0x00, 0x02, // dst ip - 0x00, 0x01, // src_port - 0x00, 0x02, // dst_port - 0x00, 0x00, 0x00, 0x00, // sequence - 0x00, 0x00, 0x00, 0x00, // ack_number - 0x00, 0x00, // tcp_header_length - 0x00, 0x00, // tcp_window_size - 0x00, 0x00, // checksum - 0x00, 0x00, // tcp_urgent_pointer - }, - ) - dc = newDecoder() - actual, err := dc.decodeIPv4Header(octets) - require.NoError(t, err) - - expected := ipV4Header{ - version: 64, - internetHeaderLength: 5, - protocol: 6, - sourceIP: [4]byte{127, 0, 0, 1}, - destIP: [4]byte{127, 0, 0, 2}, - protocolHeader: tcpHeader{ - sourcePort: 1, - destinationPort: 2, - }, - } - - require.Equal(t, expected, actual) -} - -func TestUnknownProtocol(t *testing.T) { - octets := bytes.NewBuffer( - []byte{ - 0x45, // version + IHL - 0x00, // ip_dscp + ip_ecn - 0x00, 0x00, // total length - 0x00, 0x00, // identification - 0x00, 0x00, // flags + frag offset - 0x00, // ttl - 0x99, // protocol - 0x00, 0x00, // header checksum - 0x7f, 0x00, 0x00, 0x01, // src ip - 0x7f, 0x00, 0x00, 0x02, // dst ip - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - }, - ) - dc := newDecoder() - actual, err := dc.decodeIPv4Header(octets) - require.NoError(t, err) - - expected := ipV4Header{ - version: 64, - internetHeaderLength: 5, - protocol: 153, - sourceIP: [4]byte{127, 0, 0, 1}, - destIP: [4]byte{127, 0, 0, 2}, - } - - require.Equal(t, expected, actual) -} diff --git a/plugins/inputs/sflow/sample.conf b/plugins/inputs/sflow/sample.conf deleted file mode 100644 index 8935fa3f2966e..0000000000000 --- a/plugins/inputs/sflow/sample.conf +++ /dev/null @@ -1,11 +0,0 @@ -# SFlow V5 Protocol Listener -[[inputs.sflow]] - ## Address to listen for sFlow packets. - ## example: service_address = "udp://:6343" - ## service_address = "udp4://:6343" - ## service_address = "udp6://:6343" - service_address = "udp://:6343" - - ## Set the size of the operating system's receive buffer. - ## example: read_buffer_size = "64KiB" - # read_buffer_size = "" diff --git a/plugins/inputs/sflow/sflow.go b/plugins/inputs/sflow/sflow.go deleted file mode 100644 index b703d78c80dd7..0000000000000 --- a/plugins/inputs/sflow/sflow.go +++ /dev/null @@ -1,139 +0,0 @@ -//go:generate ../../../tools/readme_config_includer/generator -package sflow - -import ( - "bytes" - _ "embed" - "fmt" - "io" - "net" - "net/url" - "strings" - "sync" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/config" - "github.com/influxdata/telegraf/plugins/inputs" -) - -//go:embed sample.conf -var sampleConfig string - -const ( - maxPacketSize = 64 * 1024 -) - -type SFlow struct { - ServiceAddress string `toml:"service_address"` - ReadBufferSize config.Size `toml:"read_buffer_size"` - - Log telegraf.Logger `toml:"-"` - - addr net.Addr - decoder *packetDecoder - closer io.Closer - wg sync.WaitGroup -} - -func (*SFlow) SampleConfig() string { - return sampleConfig -} - -func (s *SFlow) Init() error { - s.decoder = newDecoder() - s.decoder.Log = s.Log - return nil -} - -// Start starts this sFlow listener listening on the configured network for sFlow packets -func (s *SFlow) Start(acc telegraf.Accumulator) error { - s.decoder.onPacket(func(p *v5Format) { - metrics := makeMetrics(p) - for _, m := range metrics { - acc.AddMetric(m) - } - }) - - u, err := url.Parse(s.ServiceAddress) - if err != nil { - return err - } - - conn, err := listenUDP(u.Scheme, u.Host) - if err != nil { - return err - } - s.closer = conn - s.addr = conn.LocalAddr() - - if s.ReadBufferSize > 0 { - if err := conn.SetReadBuffer(int(s.ReadBufferSize)); err != nil { - return err - } - } - - s.Log.Infof("Listening on %s://%s", s.addr.Network(), s.addr.String()) - - s.wg.Add(1) - go func() { - defer s.wg.Done() - s.read(acc, conn) - }() - - return nil -} - -// Gather is a NOOP for sFlow as it receives, asynchronously, sFlow network packets -func (*SFlow) Gather(telegraf.Accumulator) error { - return nil -} - -func (s *SFlow) Stop() { - if s.closer != nil { - s.closer.Close() - } - s.wg.Wait() -} - -func (s *SFlow) address() net.Addr { - return s.addr -} - -func (s *SFlow) read(acc telegraf.Accumulator, conn net.PacketConn) { - buf := make([]byte, maxPacketSize) - for { - n, _, err := conn.ReadFrom(buf) - if err != nil { - if !strings.HasSuffix(err.Error(), ": use of closed network connection") { - acc.AddError(err) - } - break - } - s.process(acc, buf[:n]) - } -} - -func (s *SFlow) process(acc telegraf.Accumulator, buf []byte) { - if err := s.decoder.decode(bytes.NewBuffer(buf)); err != nil { - acc.AddError(fmt.Errorf("unable to parse incoming packet: %w", err)) - } -} - -func listenUDP(network, address string) (*net.UDPConn, error) { - switch network { - case "udp", "udp4", "udp6": - addr, err := net.ResolveUDPAddr(network, address) - if err != nil { - return nil, err - } - return net.ListenUDP(network, addr) - default: - return nil, fmt.Errorf("unsupported network type: %s", network) - } -} - -func init() { - inputs.Add("sflow", func() telegraf.Input { - return &SFlow{} - }) -} diff --git a/plugins/inputs/sflow/sflow_test.go b/plugins/inputs/sflow/sflow_test.go deleted file mode 100644 index e6d1e73d3758f..0000000000000 --- a/plugins/inputs/sflow/sflow_test.go +++ /dev/null @@ -1,157 +0,0 @@ -package sflow - -import ( - "encoding/hex" - "net" - "testing" - "time" - - "github.com/stretchr/testify/require" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/metric" - "github.com/influxdata/telegraf/testutil" -) - -func TestSFlow(t *testing.T) { - sflow := &SFlow{ - ServiceAddress: "udp://127.0.0.1:0", - Log: testutil.Logger{}, - } - err := sflow.Init() - require.NoError(t, err) - - var acc testutil.Accumulator - err = sflow.Start(&acc) - require.NoError(t, err) - defer sflow.Stop() - - client, err := net.Dial(sflow.address().Network(), sflow.address().String()) - require.NoError(t, err) - - packetBytes, err := hex.DecodeString( - "0000000500000001c0a80102000000100000f3d40bfa047f0000000200000001000000d00001210a000001fe000004000484240000000000000001fe" + - "00000200000000020000000100000090000000010000010b0000000400000080000c2936d3d694c691aa97600800450000f9f19040004011b4f5c0a" + - "80913c0a8090a00a1ba0500e5641f3081da02010104066d6f746f6770a281cc02047b46462e0201000201003081bd3012060d2b0601020119050101" + - "0281dc710201003013060d2b06010201190501010281e66802025acc3012060d2b0601020119050101000003e900000010000000090000000000000" + - "0090000000000000001000000d00000e3cc000002100000400048eb7400000000000000021000000200000000020000000100000090000000010000" + - "00970000000400000080000c2936d3d6fcecda44008f81000009080045000081186440003f119098c0a80815c0a8090a9a690202006d23083c33303" + - "e4170722031312030393a33333a3031206b6e6f64653120736e6d70645b313039385d3a20436f6e6e656374696f6e2066726f6d205544503a205b31" + - "39322e3136382e392e31305d3a34393233362d000003e90000001000000009000000000000000900000000", - ) - require.NoError(t, err) - _, err = client.Write(packetBytes) - require.NoError(t, err) - - acc.Wait(2) - - expected := []telegraf.Metric{ - metric.New( - "sflow", - map[string]string{ - "agent_address": "192.168.1.2", - "dst_ip": "192.168.9.10", - "dst_mac": "00:0c:29:36:d3:d6", - "dst_port": "47621", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "510", - "output_ifindex": "512", - "sample_direction": "ingress", - "source_id_index": "510", - "source_id_type": "0", - "src_ip": "192.168.9.19", - "src_mac": "94:c6:91:aa:97:60", - "src_port": "161", - }, - map[string]interface{}{ - "bytes": uint64(273408), - "drops": uint64(0), - "frame_length": uint64(267), - "header_length": uint64(128), - "ip_flags": uint64(2), - "ip_fragment_offset": uint64(0), - "ip_total_length": uint64(249), - "ip_ttl": uint64(64), - "sampling_rate": uint64(1024), - "udp_length": uint64(229), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - metric.New( - "sflow", - map[string]string{ - "agent_address": "192.168.1.2", - "dst_ip": "192.168.9.10", - "dst_mac": "00:0c:29:36:d3:d6", - "dst_port": "514", - "ether_type": "IPv4", - "header_protocol": "ETHERNET-ISO88023", - "input_ifindex": "528", - "output_ifindex": "512", - "sample_direction": "ingress", - "source_id_index": "528", - "source_id_type": "0", - "src_ip": "192.168.8.21", - "src_mac": "fc:ec:da:44:00:8f", - "src_port": "39529", - }, - map[string]interface{}{ - "bytes": uint64(2473984), - "drops": uint64(0), - "frame_length": uint64(151), - "header_length": uint64(128), - "ip_flags": uint64(2), - "ip_fragment_offset": uint64(0), - "ip_total_length": uint64(129), - "ip_ttl": uint64(63), - "sampling_rate": uint64(16384), - "udp_length": uint64(109), - "ip_dscp": "0", - "ip_ecn": "0", - }, - time.Unix(0, 0), - ), - } - - testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), - testutil.IgnoreTime()) -} - -func BenchmarkSFlow(b *testing.B) { - sflow := &SFlow{ - ServiceAddress: "udp://127.0.0.1:0", - Log: testutil.Logger{}, - } - err := sflow.Init() - require.NoError(b, err) - - var acc testutil.Accumulator - err = sflow.Start(&acc) - require.NoError(b, err) - defer sflow.Stop() - - client, err := net.Dial(sflow.address().Network(), sflow.address().String()) - require.NoError(b, err) - - packetBytes, err := hex.DecodeString( - "0000000500000001c0a80102000000100000f3d40bfa047f0000000200000001000000d00001210a000001fe000004000484240000000000000001fe0" + - "0000200000000020000000100000090000000010000010b0000000400000080000c2936d3d694c691aa97600800450000f9f19040004011b4f5c0a80" + - "913c0a8090a00a1ba0500e5641f3081da02010104066d6f746f6770a281cc02047b46462e0201000201003081bd3012060d2b0601020119050101028" + - "1dc710201003013060d2b06010201190501010281e66802025acc3012060d2b0601020119050101000003e9000000100000000900000000000000090" + - "000000000000001000000d00000e3cc000002100000400048eb740000000000000002100000020000000002000000010000009000000001000000970" + - "000000400000080000c2936d3d6fcecda44008f81000009080045000081186440003f119098c0a80815c0a8090a9a690202006d23083c33303e41707" + - "22031312030393a33333a3031206b6e6f64653120736e6d70645b313039385d3a20436f6e6e656374696f6e2066726f6d205544503a205b3139322e3" + - "136382e392e31305d3a34393233362d000003e90000001000000009000000000000000900000000", - ) - require.NoError(b, err) - - b.ResetTimer() - for n := 0; n < b.N; n++ { - _, err := client.Write(packetBytes) - require.NoError(b, err) - acc.Wait(2) - } -} diff --git a/plugins/inputs/sflow/types.go b/plugins/inputs/sflow/types.go deleted file mode 100644 index ce4374ef99c38..0000000000000 --- a/plugins/inputs/sflow/types.go +++ /dev/null @@ -1,297 +0,0 @@ -package sflow - -import ( - "net" - "strconv" -) - -const ( - ipProtocolTCP uint8 = 6 - ipProtocolUDP uint8 = 17 -) - -var eTypeMap = map[uint16]string{ - 0x0800: "IPv4", - 0x86DD: "IPv6", -} - -type containsMetricData interface { - getTags() map[string]string - getFields() map[string]interface{} -} - -// v5Format answers and decoder.Directive capable of decoding sFlow v5 packets in accordance -// with SFlow v5 specification at https://sflow.org/sflow_version_5.txt -type v5Format struct { - version uint32 - agentAddress net.IPAddr - subAgentID uint32 - sequenceNumber uint32 - uptime uint32 - samples []sample -} - -type sampleType uint32 - -const ( - sampleTypeFlowSample sampleType = 1 // sflow_version_5.txt line: 1614 - sampleTypeFlowSampleExpanded sampleType = 3 // sflow_version_5.txt line: 1698 -) - -type sample struct { - smplType sampleType - smplData sampleDataFlowSampleExpanded -} - -type sampleDataFlowSampleExpanded struct { - sequenceNumber uint32 - sourceIDType uint32 - sourceIDIndex uint32 - samplingRate uint32 - samplePool uint32 - drops uint32 - sampleDirection string // ingress/egress - inputIfFormat uint32 - inputIfIndex uint32 - outputIfFormat uint32 - outputIfIndex uint32 - flowRecords []flowRecord -} - -type flowFormatType uint32 - -const ( - flowFormatTypeRawPacketHeader flowFormatType = 1 // sflow_version_5.txt line: 1938 -) - -type flowData containsMetricData - -type flowRecord struct { - flowFormat flowFormatType - flowData flowData -} - -type headerProtocolType uint32 - -const ( - headerProtocolTypeEthernetISO88023 headerProtocolType = 1 - headerProtocolTypeISO88024TokenBus headerProtocolType = 2 - headerProtocolTypeISO88025TokenRing headerProtocolType = 3 - headerProtocolTypeFDDI headerProtocolType = 4 - headerProtocolTypeFrameRelay headerProtocolType = 5 - headerProtocolTypeX25 headerProtocolType = 6 - headerProtocolTypePPP headerProtocolType = 7 - headerProtocolTypeSMDS headerProtocolType = 8 - headerProtocolTypeAAL5 headerProtocolType = 9 - headerProtocolTypeAAL5IP headerProtocolType = 10 /* e.g. Cisco AAL5 mux */ - headerProtocolTypeIPv4 headerProtocolType = 11 - headerProtocolTypeIPv6 headerProtocolType = 12 - headerProtocolTypeMPLS headerProtocolType = 13 - headerProtocolTypePOS headerProtocolType = 14 /* RFC 1662, 2615 */ -) - -var headerProtocolMap = map[headerProtocolType]string{ - headerProtocolTypeEthernetISO88023: "ETHERNET-ISO88023", // sflow_version_5.txt line: 1920 -} - -type header containsMetricData - -type rawPacketHeaderFlowData struct { - headerProtocol headerProtocolType - frameLength uint32 - bytes uint32 - strippedOctets uint32 - headerLength uint32 - header header -} - -func (h rawPacketHeaderFlowData) getTags() map[string]string { - var t map[string]string - if h.header != nil { - t = h.header.getTags() - } else { - t = make(map[string]string, 1) - } - t["header_protocol"] = headerProtocolMap[h.headerProtocol] - return t -} - -func (h rawPacketHeaderFlowData) getFields() map[string]interface{} { - var f map[string]interface{} - if h.header != nil { - f = h.header.getFields() - } else { - f = make(map[string]interface{}, 3) - } - f["bytes"] = h.bytes - f["frame_length"] = h.frameLength - f["header_length"] = h.headerLength - return f -} - -type ipHeader containsMetricData - -type ethHeader struct { - destinationMAC [6]byte - sourceMAC [6]byte - tagProtocolIdentifier uint16 - tagControlInformation uint16 - etherTypeCode uint16 - etherType string - ipHeader ipHeader -} - -func (h ethHeader) getTags() map[string]string { - var t map[string]string - if h.ipHeader != nil { - t = h.ipHeader.getTags() - } else { - t = make(map[string]string, 3) - } - t["src_mac"] = net.HardwareAddr(h.sourceMAC[:]).String() - t["dst_mac"] = net.HardwareAddr(h.destinationMAC[:]).String() - t["ether_type"] = h.etherType - return t -} - -func (h ethHeader) getFields() map[string]interface{} { - if h.ipHeader != nil { - return h.ipHeader.getFields() - } - return make(map[string]interface{}) -} - -type protocolHeader containsMetricData - -// https://en.wikipedia.org/wiki/IPv4#Header -type ipV4Header struct { - version uint8 // 4 bit - internetHeaderLength uint8 // 4 bit - dscp uint8 - ecn uint8 - totalLength uint16 - identification uint16 - flags uint8 - fragmentOffset uint16 - ttl uint8 - protocol uint8 // https://en.wikipedia.org/wiki/List_of_IP_protocol_numbers - headerChecksum uint16 - sourceIP [4]byte - destIP [4]byte - protocolHeader protocolHeader -} - -func (h ipV4Header) getTags() map[string]string { - var t map[string]string - if h.protocolHeader != nil { - t = h.protocolHeader.getTags() - } else { - t = make(map[string]string, 2) - } - t["src_ip"] = net.IP(h.sourceIP[:]).String() - t["dst_ip"] = net.IP(h.destIP[:]).String() - return t -} - -func (h ipV4Header) getFields() map[string]interface{} { - var f map[string]interface{} - if h.protocolHeader != nil { - f = h.protocolHeader.getFields() - } else { - f = make(map[string]interface{}, 6) - } - f["ip_dscp"] = strconv.FormatUint(uint64(h.dscp), 10) - f["ip_ecn"] = strconv.FormatUint(uint64(h.ecn), 10) - f["ip_flags"] = h.flags - f["ip_fragment_offset"] = h.fragmentOffset - f["ip_total_length"] = h.totalLength - f["ip_ttl"] = h.ttl - return f -} - -// https://en.wikipedia.org/wiki/IPv6_packet -type ipV6Header struct { - dscp uint8 - ecn uint8 - payloadLength uint16 - nextHeaderProto uint8 // tcp/udp? - hopLimit uint8 - sourceIP [16]byte - destIP [16]byte - protocolHeader protocolHeader -} - -func (h ipV6Header) getTags() map[string]string { - var t map[string]string - if h.protocolHeader != nil { - t = h.protocolHeader.getTags() - } else { - t = make(map[string]string, 2) - } - t["src_ip"] = net.IP(h.sourceIP[:]).String() - t["dst_ip"] = net.IP(h.destIP[:]).String() - return t -} - -func (h ipV6Header) getFields() map[string]interface{} { - var f map[string]interface{} - if h.protocolHeader != nil { - f = h.protocolHeader.getFields() - } else { - f = make(map[string]interface{}, 3) - } - f["ip_dscp"] = strconv.FormatUint(uint64(h.dscp), 10) - f["ip_ecn"] = strconv.FormatUint(uint64(h.ecn), 10) - f["payload_length"] = h.payloadLength - return f -} - -// https://en.wikipedia.org/wiki/Transmission_Control_Protocol -type tcpHeader struct { - sourcePort uint16 - destinationPort uint16 - sequence uint32 - ackNumber uint32 - tcpHeaderLength uint8 - flags uint16 - tcpWindowSize uint16 - checksum uint16 - tcpUrgentPointer uint16 -} - -func (h tcpHeader) getTags() map[string]string { - t := map[string]string{ - "dst_port": strconv.FormatUint(uint64(h.destinationPort), 10), - "src_port": strconv.FormatUint(uint64(h.sourcePort), 10), - } - return t -} - -func (h tcpHeader) getFields() map[string]interface{} { - return map[string]interface{}{ - "tcp_header_length": h.tcpHeaderLength, - "tcp_urgent_pointer": h.tcpUrgentPointer, - "tcp_window_size": h.tcpWindowSize, - } -} - -type udpHeader struct { - sourcePort uint16 - destinationPort uint16 - udpLength uint16 - checksum uint16 -} - -func (h udpHeader) getTags() map[string]string { - t := map[string]string{ - "dst_port": strconv.FormatUint(uint64(h.destinationPort), 10), - "src_port": strconv.FormatUint(uint64(h.sourcePort), 10), - } - return t -} - -func (h udpHeader) getFields() map[string]interface{} { - return map[string]interface{}{ - "udp_length": h.udpLength, - } -} diff --git a/plugins/inputs/sflow/types_test.go b/plugins/inputs/sflow/types_test.go deleted file mode 100644 index 614a2e092539b..0000000000000 --- a/plugins/inputs/sflow/types_test.go +++ /dev/null @@ -1,43 +0,0 @@ -package sflow - -import ( - "testing" - - "github.com/stretchr/testify/require" -) - -func TestRawPacketHeaderFlowData(t *testing.T) { - h := rawPacketHeaderFlowData{ - headerProtocol: headerProtocolTypeEthernetISO88023, - frameLength: 64, - bytes: 64, - strippedOctets: 0, - headerLength: 0, - header: nil, - } - tags := h.getTags() - fields := h.getFields() - - require.NotNil(t, fields) - require.NotNil(t, tags) - require.Contains(t, tags, "header_protocol") - require.Len(t, tags, 1) -} - -// process a raw ethernet packet without any encapsulated protocol -func TestEthHeader(t *testing.T) { - h := ethHeader{ - destinationMAC: [6]byte{0xca, 0xff, 0xee, 0xff, 0xe, 0x0}, - sourceMAC: [6]byte{0xde, 0xad, 0xbe, 0xef, 0x0, 0x0}, - tagProtocolIdentifier: 0x88B5, // IEEE Std 802 - Local Experimental Ethertype - tagControlInformation: 0, - etherTypeCode: 0, - etherType: "", - ipHeader: nil, - } - tags := h.getTags() - fields := h.getFields() - - require.NotNil(t, fields) - require.NotNil(t, tags) -}