Repository: kentik/ktranslate
Pull request #539 (2023-03-08 23:29:11)

Bump golang.org/x/text from 0.3.7 to 0.3.8
Bumps [golang.org/x/text](https://github.com/golang/text) from 0.3.7 to 0.3.8.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/golang/text/commit/434eadcdbc3b0256971992e8c70027278364c72c"><code>434eadc</code></a> language: reject excessively large Accept-Language strings</li>
<li><a href="https://github.com/golang/text/commit/23407e72ed5b895a2dfd230aec777f4fbe026d6a"><code>23407e7</code></a> go.mod: ignore cyclic dependency for tagging</li>
<li><a href="https://github.com/golang/text/commit/b18d3dd8a4b426ebedcf279b593e85ac4985b9d3"><code>b18d3dd</code></a> secure/precis: replace bytes.Compare with bytes.Equal</li>
<li><a href="https://github.com/golang/text/commit/795e854ff348c9cac4fd0033ce04c417705dd0bb"><code>795e854</code></a> all: replace io/ioutil with io and os package</li>
<li><a href="https://github.com/golang/text/commit/b0ca10ff35f1325c7d0ac7830fe3f036bd72d8f9"><code>b0ca10f</code></a> internal/language: bump script types to uint16 and update registry</li>
<li><a href="https://github.com/golang/text/commit/ba9b0e1d4b03523c708709935fbc961124b6967b"><code>ba9b0e1</code></a> go.mod: update x/tools to HEAD</li>
<li><a href="https://github.com/golang/text/commit/d03b41800055b01e3895b1e047af09733c93bf63"><code>d03b418</code></a> A+C: delete AUTHORS and CONTRIBUTORS</li>
<li><a href="https://github.com/golang/text/commit/b4bca84b03619dba00657375259024a7f8ae6712"><code>b4bca84</code></a> language/display: fix Tag method comment</li>
<li><a href="https://github.com/golang/text/commit/ea49e3e2d5b3f1518081d8bc53ffefc8bc60ecec"><code>ea49e3e</code></a> go.mod: update x/tools to HEAD</li>
<li><a href="https://github.com/golang/text/commit/78819d01d041a94e055bbaa2d95e5e4d49e8f8a0"><code>78819d0</code></a> go.mod: update to golang.org/x/text v0.1.10</li>
<li>Additional commits viewable in <a href="https://github.com/golang/text/compare/v0.3.7...v0.3.8">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=golang.org/x/text&package-manager=go_modules&previous-version=0.3.7&new-version=0.3.8)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
- `@dependabot use these labels` will set the current labels as the default for future PRs for this repo and language
- `@dependabot use these reviewers` will set the current reviewers as the default for future PRs for this repo and language
- `@dependabot use these assignees` will set the current assignees as the default for future PRs for this repo and language
- `@dependabot use this milestone` will set the current milestone as the default for future PRs for this repo and language

You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/kentik/ktranslate/network/alerts).

</details>


Patch:
diff --git a/.github/workflows/publish-kentik.yml b/.github/workflows/publish-kentik.yml
index c53bb08..6b42e1d 100644
--- a/.github/workflows/publish-kentik.yml
+++ b/.github/workflows/publish-kentik.yml
@@ -1,4 +1,5 @@
 name: Publish Kentik
+run-name: ${{ inputs.message }}
 
 on:
   push:
@@ -6,6 +7,9 @@ on:
       - kentik
 
   workflow_dispatch:
+    inputs:
+      message:
+        type: string
 
 jobs:
   deploy:
diff --git a/README.md b/README.md
index 7d2f93a..b7c6e8e 100644
--- a/README.md
+++ b/README.md
@@ -140,7 +140,7 @@ To get your own MaxMind key, visit [MaxMind](https://www.maxmind.com).
   -nr_estimate_only
     	If true, record size of inputs to NR but don't actually send anything
   -nr_region string
-    	NR Region to use. US|EU
+       NR Region to use. US|EU|GOV
   -olly_dataset string
     	Olly dataset name
   -olly_write_key string
diff --git a/config.go b/config.go
index ee6399f..8228d7c 100644
--- a/config.go
+++ b/config.go
@@ -13,6 +13,9 @@ const (
 	FlowDefaultFields = "TimeReceived,SamplingRate,Bytes,Packets,SrcAddr,DstAddr,Proto,SrcPort,DstPort,InIf,OutIf,SrcVlan,DstVlan,TCPFlags,SrcAS,DstAS,Type,SamplerAddress,FlowDirection"
 	// KentikAPITokenEnvVar is the environment variables used to get the Kentik API Token
 	KentikAPITokenEnvVar = "KENTIK_API_TOKEN"
+	// MaxNetflowsPerMessage is the maximum flowsets in a single netflow message
+	// Must be less than ( 65535[nf_max_size] - 16[nf_header_size] - 96(nf_template_flowset_size] ) / 128[nf_data_flowset_size]
+	MaxNetflowsPerMessage = 510
 )
 
 // NetflowFormatConfig is the config format for netflow
@@ -494,6 +497,10 @@ func LoadConfig(configPath string) (*Config, error) {
 		return nil, err
 	}
 
+	if cfg.Format == "netflow" && (cfg.MaxFlowsPerMessage <= 0 || cfg.MaxFlowsPerMessage > MaxNetflowsPerMessage) {
+		cfg.MaxFlowsPerMessage = MaxNetflowsPerMessage
+	}
+
 	return cfg, nil
 }
 
diff --git a/go.mod b/go.mod
index 710ea52..ea2ffd6 100644
--- a/go.mod
+++ b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/aws/aws-sdk-go v1.41.19
 	github.com/dchest/siphash v1.2.2
 	github.com/elliotchance/orderedmap v1.4.0
+	github.com/fsnotify/fsnotify v1.6.0
 	github.com/go-openapi/runtime v0.24.1
 	github.com/go-ping/ping v0.0.0-20211014180314-6e2b003bffdd
 	github.com/google/uuid v1.3.0
@@ -67,7 +68,6 @@ require (
 	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect
 	github.com/facebookgo/limitgroup v0.0.0-20150612190941-6abd8d71ec01 // indirect
 	github.com/facebookgo/muster v0.0.0-20150708232844-fd3d7953fd52 // indirect
-	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/go-openapi/analysis v0.21.2 // indirect
 	github.com/go-openapi/errors v0.20.2 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
@@ -123,7 +123,7 @@ require (
 	golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/sys v0.2.0 // indirect
-	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/text v0.3.8 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/api v0.74.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
diff --git a/go.sum b/go.sum
index 57e02b0..3859264 100644
--- a/go.sum
+++ b/go.sum
@@ -169,8 +169,6 @@ github.com/frankban/quicktest v1.13.0 h1:yNZif1OkDfNoDfb9zZa9aXIpejNR4F23Wely0c+
 github.com/frankban/quicktest v1.13.0/go.mod h1:qLE0fzW0VuyUAJgPU19zByoIr0HtCHN/r/VLSOOIySU=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
-github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
-github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -797,7 +795,6 @@ golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
@@ -814,8 +811,9 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/pkg/formats/netflow/netflow.go b/pkg/formats/netflow/netflow.go
index cf43fe6..518ea6c 100644
--- a/pkg/formats/netflow/netflow.go
+++ b/pkg/formats/netflow/netflow.go
@@ -58,6 +58,7 @@ const (
 	IPV6_DST_ADDR             = 28
 	SRC_MAC                   = 56
 	DST_MAC                   = 80
+	OBSERVATION_POINT_ID      = 138
 	FLOW_START_SECONDS        = 150
 	FLOW_END_SECONDS          = 151
 )
@@ -92,6 +93,7 @@ var fields = []Field{
 	{OUT_BYTES, 8},
 	{IN_PKTS, 8},
 	{OUT_PKTS, 8},
+	{OBSERVATION_POINT_ID, 8}, // uint64
 	{FLOW_START_SECONDS, 4}, // uint32
 	{FLOW_END_SECONDS, 4},
 }
@@ -434,6 +436,8 @@ func encodeFlow(flow *kt.JCHF) (*bytes.Buffer, error) {
 			value = uint64(flow.InPkts)
 		case OUT_PKTS:
 			value = uint64(flow.OutPkts)
+		case OBSERVATION_POINT_ID:
+			value = uint64(flow.CompanyId&0xffff)<<32 + uint64(flow.DeviceId)
 		case FLOW_START_SECONDS:
 			value = uint32(flow.Timestamp)
 		case FLOW_END_SECONDS: // Assume that each flow is 60 sec long.
diff --git a/pkg/inputs/snmp/metadata/poll.go b/pkg/inputs/snmp/metadata/poll.go
index 12d3caa..f00b2a0 100644
--- a/pkg/inputs/snmp/metadata/poll.go
+++ b/pkg/inputs/snmp/metadata/poll.go
@@ -260,8 +260,9 @@ func (p *Poller) toFlows(dd *kt.DeviceData) ([]*kt.JCHF, error) {
 	for k, v := range cs {
 		if strings.HasPrefix(v, TagValuePrefix) { // See if we can find this value in the MDS instead.
 			for tag, value := range dst.CustomStr {
-				if tag == v[len(TagValuePrefix):] {
+				if len(v) > len(TagValuePrefix) && tag == v[len(TagValuePrefix):] {
 					v = value
+					break // Once we find a match, break out and don't try comparing to anything else any more.
 				}
 			}
 		}
diff --git a/pkg/inputs/snmp/metrics/poll.go b/pkg/inputs/snmp/metrics/poll.go
index dda9fc1..32ac54a 100644
--- a/pkg/inputs/snmp/metrics/poll.go
+++ b/pkg/inputs/snmp/metrics/poll.go
@@ -26,6 +26,7 @@ type Poller struct {
 	jchfChan         chan []*kt.JCHF
 	metrics          *kt.SnmpDeviceMetric
 	counterTimeSec   int
+	jitterTimeSec    int
 	dropIfOutside    bool
 	pinger           *ping.Pinger
 	extension        extension.Extension
@@ -45,6 +46,11 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 		counterTimeSec = 30
 	}
 
+	jitterTimeSec := 15 // This is how long to spead the polling load out across.
+	if gconf.JitterTimeSec > 0 {
+		jitterTimeSec = gconf.JitterTimeSec
+	}
+
 	// Default is not not drop.
 	dropIfOutside := false
 	if gconf != nil && gconf.PollTimeSec > 0 {
@@ -70,6 +76,7 @@ func NewPoller(server *gosnmp.GoSNMP, gconf *kt.SnmpGlobalConfig, conf *kt.SnmpD
 		interfaceMetrics: NewInterfaceMetrics(gconf, conf, metrics, interfaceMetricMibs, profile, counterTimeSec, log),
 		deviceMetrics:    NewDeviceMetrics(gconf, conf, metrics, deviceMetricMibs, profile, log),
 		counterTimeSec:   counterTimeSec,
+		jitterTimeSec:    jitterTimeSec,
 		dropIfOutside:    dropIfOutside,
 	}
 
@@ -99,6 +106,11 @@ func NewPollerForPing(gconf *kt.SnmpGlobalConfig, conf *kt.SnmpDeviceConfig, jch
 		counterTimeSec = 30
 	}
 
+	jitterTimeSec := 15 // This is how long to spead the polling load out across.
+	if gconf.JitterTimeSec > 0 {
+		jitterTimeSec = gconf.JitterTimeSec
+	}
+
 	pingSec := conf.PingSec
 	if pingSec == 0 { // If not per device, try per global.
 		pingSec = gconf.PingSec
@@ -112,6 +124,7 @@ func NewPollerForPing(gconf *kt.SnmpGlobalConfig, conf *kt.SnmpDeviceConfig, jch
 		log:            log,
 		metrics:        metrics,
 		counterTimeSec: counterTimeSec,
+		jitterTimeSec:  jitterTimeSec,
 		deviceMetrics:  NewDeviceMetrics(gconf, conf, metrics, nil, profile, log),
 	}
 
@@ -140,11 +153,17 @@ func NewPollerForExtention(gconf *kt.SnmpGlobalConfig, conf *kt.SnmpDeviceConfig
 		counterTimeSec = 30
 	}
 
+	jitterTimeSec := 15 // This is how long to spead the polling load out across.
+	if gconf.JitterTimeSec > 0 {
+		jitterTimeSec = gconf.JitterTimeSec
+	}
+
 	poller := Poller{
 		jchfChan:       jchfChan,
 		log:            log,
 		metrics:        metrics,
 		counterTimeSec: counterTimeSec,
+		jitterTimeSec:  jitterTimeSec,
 		deviceMetrics:  NewDeviceMetrics(gconf, conf, metrics, nil, profile, log),
 	}
 
@@ -167,7 +186,7 @@ func (p *Poller) StartLoop(ctx context.Context) {
 	// to jitter the devices, and the rest of the five-minute chunk to actually do the counter-polling.  For any device whose counters we can walk
 	// in less than (5 minutes - jitter period), we should be able to guarantee exactly one datapoint per aligned five-minute chunk.
 	counterAlignment := time.Duration(p.counterTimeSec) * time.Second
-	jitterWindow := 15 * time.Second
+	jitterWindow := time.Duration(p.jitterTimeSec) * time.Second
 	firstCollection := time.Now().Truncate(counterAlignment).Add(counterAlignment).Add(time.Duration(rand.Int63n(int64(jitterWindow))))
 	counterCheck := tick.NewFixedTimer(firstCollection, counterAlignment)
 	statusCheck := time.NewTicker(STATUS_CHECK_TIME)
@@ -278,7 +297,7 @@ func (p *Poller) StartPingOnlyLoop(ctx context.Context) {
 	// to jitter the devices, and the rest of the five-minute chunk to actually do the counter-polling.  For any device whose counters we can walk
 	// in less than (5 minutes - jitter period), we should be able to guarantee exactly one datapoint per aligned five-minute chunk.
 	counterAlignment := time.Duration(p.counterTimeSec) * time.Second
-	jitterWindow := 15 * time.Second
+	jitterWindow := time.Duration(p.jitterTimeSec) * time.Second
 	firstCollection := time.Now().Truncate(counterAlignment).Add(counterAlignment).Add(time.Duration(rand.Int63n(int64(jitterWindow))))
 	counterCheck := tick.NewFixedTimer(firstCollection, counterAlignment)
 	p.deviceMetrics.ResetPingStats() // Initialize to 0 sent and recieved.
diff --git a/pkg/kt/snmp.go b/pkg/kt/snmp.go
index 08484c3..23ea393 100644
--- a/pkg/kt/snmp.go
+++ b/pkg/kt/snmp.go
@@ -270,6 +270,7 @@ type SnmpGlobalConfig struct {
 	UserTags              map[string]string      `yaml:"user_tags"`
 	MatchAttr             map[string]string      `yaml:"match_attributes"`
 	ProviderMap           map[string]ProviderMap `yaml:"providers"`
+	JitterTimeSec         int                    `yaml:"jitter_time_sec"`
 }
 
 type SnmpConfig struct {


Test patch:
diff --git a/pkg/inputs/snmp/metadata/poll_test.go b/pkg/inputs/snmp/metadata/poll_test.go
index 8c71f43..22d2a47 100644
--- a/pkg/inputs/snmp/metadata/poll_test.go
+++ b/pkg/inputs/snmp/metadata/poll_test.go
@@ -40,3 +40,37 @@ func TestGetDeviceManufacturer(t *testing.T) {
 		}
 	}
 }
+
+func TestToFlows(t *testing.T) {
+	l := lt.NewTestContextL(logger.NilContext, t)
+	conf := &kt.SnmpDeviceConfig{
+		Provider: "foo",
+		UserTags: map[string]string{
+			"foo": "$foo",
+			"aaa": "$SysContact",
+		},
+	}
+	conf.InitUserTags("service")
+
+	p := &Poller{
+		log:  l,
+		conf: conf,
+	}
+
+	input := kt.DeviceData{
+		Manufacturer: "man",
+		DeviceMetricsMetadata: &kt.DeviceMetricsMetadata{
+			SysContact: "ddd",
+			Customs: map[string]string{
+				"foo": "",
+				"bar": "",
+			},
+		},
+	}
+	res, err := p.toFlows(&input)
+	assert.NotNil(t, res)
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(res))
+	assert.Equal(t, "ddd", res[0].CustomStr["tags.aaa"])
+	assert.Equal(t, "", res[0].CustomStr["tags.foo"]) // Empty string match for tag here.
+}