Skip to content

Commit d6ed56b

Browse files
i3149Copilot
i3149
and
Copilot
authored
Flow stitch basic (#902)
* Basic idea for a flow stitching system * Adding tests for stitch * removing log line and verifying delete works * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Better token for get key * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Adding in some metrics for flow stitching * fixing test * Adding a test for flatten and also code to make the pair flatten work * defaulting flows kept to 1 sec * Adding in a cached version of rollup * getting there with a ringbuffer * Working poc for flow stitching * del after finding a match for now * fixing test * tidying the go mod file * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * cleaning up with robot suggestions --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent c0164f8 commit d6ed56b

14 files changed

Lines changed: 1190 additions & 8 deletions

File tree

‎cmd/ktranslate/main.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -910,6 +910,20 @@ func applyFlags(cfg *ktranslate.Config) error {
910910
cfg.FlowInput.MappingFile = val
911911
case "config_provider":
912912
cfg.CfgManager.ConfigImpl = val
913+
case "stitch.buffer.len":
914+
v, err := strconv.Atoi(val)
915+
if err != nil {
916+
errCh <- err
917+
return
918+
}
919+
cfg.Lilo.BufLen = v
920+
case "stitch.enable":
921+
v, err := strconv.ParseBool(val)
922+
if err != nil {
923+
errCh <- err
924+
return
925+
}
926+
cfg.Lilo.Enable = v
913927
// configs
914928
case "config", "generate-config":
915929
// ignore

‎config.go‎

Lines changed: 26 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -184,10 +184,13 @@ type DDogSinkConfig struct {
184184

185185
// RollupConfig is the config for rollups
186186
type RollupConfig struct {
187-
JoinKey string
188-
TopK int
189-
Formats []string
190-
KeepUndefined bool
187+
JoinKey string
188+
TopK int
189+
Formats []string
190+
KeepUndefined bool
191+
MaxMemoryMB int
192+
MaxKeys int
193+
EmergencyCleanup bool
191194
}
192195

193196
// KMuxConfig is the config for the mux server
@@ -285,6 +288,12 @@ type ConfigManager struct {
285288
PollTimeSec int
286289
}
287290

291+
// StitchConfig is the config on how to manage stitching flows together
292+
type StitchConfig struct {
293+
Enable bool
294+
BufLen int
295+
}
296+
288297
// Config is the ktranslate configuration
289298
type Config struct {
290299
// ktranslate
@@ -393,6 +402,8 @@ type Config struct {
393402
FlowInput *FlowInputConfig
394403
// pkg/config
395404
CfgManager *ConfigManager
405+
// pkg/stitch
406+
Lilo *StitchConfig
396407
}
397408

398409
// DefaultConfig returns a ktranslate configuration with defaults applied
@@ -543,10 +554,13 @@ func DefaultConfig() *Config {
543554
RelayURL: "",
544555
},
545556
Rollup: &RollupConfig{
546-
JoinKey: "^",
547-
TopK: 10,
548-
Formats: []string{},
549-
KeepUndefined: false,
557+
JoinKey: "^",
558+
TopK: 10,
559+
Formats: []string{},
560+
KeepUndefined: false,
561+
MaxMemoryMB: 100,
562+
MaxKeys: 5000,
563+
EmergencyCleanup: true,
550564
},
551565
KMux: &KMuxConfig{
552566
Dir: ".",
@@ -618,6 +632,10 @@ func DefaultConfig() *Config {
618632
ConfigImpl: "",
619633
PollTimeSec: 1200,
620634
},
635+
Lilo: &StitchConfig{
636+
Enable: false,
637+
BufLen: 10000,
638+
},
621639
}
622640
}
623641

‎pkg/cat/jchf.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -627,6 +627,11 @@ func (kc *KTranslate) doEnrichments(ctx context.Context, msgs []*kt.JCHF) []*kt.
627627
}
628628
}
629629
}
630+
631+
// If there's a flow stitching system enabled, try to stitch it together here.
632+
if kc.stitcher != nil {
633+
kc.stitcher.Stitch(msg)
634+
}
630635
}
631636

632637
// If there's an outside enrichment service, send over here.

‎pkg/cat/kkc.go‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/kentik/ktranslate/pkg/rollup"
2828
ss "github.com/kentik/ktranslate/pkg/sinks"
2929
"github.com/kentik/ktranslate/pkg/sinks/s3"
30+
"github.com/kentik/ktranslate/pkg/stitch"
3031
"github.com/kentik/ktranslate/pkg/util/enrich"
3132
"github.com/kentik/ktranslate/pkg/util/gopatricia/patricia"
3233
"github.com/kentik/ktranslate/pkg/util/resolv"
@@ -267,6 +268,12 @@ func NewKTranslate(config *ktranslate.Config, log logger.ContextL, registry go_m
267268
defaultProvider = kt.Provider(dp)
268269
}
269270

271+
stitcher, err := stitch.NewStitcher(log.GetLogger().GetUnderlyingLogger(), config.Lilo, registry)
272+
if err != nil {
273+
return nil, err
274+
}
275+
kc.stitcher = stitcher
276+
270277
return kc, nil
271278
}
272279

@@ -297,6 +304,9 @@ func (kc *KTranslate) cleanup() {
297304
if kc.confMgr != nil {
298305
kc.confMgr.Close()
299306
}
307+
if kc.stitcher != nil {
308+
kc.stitcher.Stop()
309+
}
300310
}
301311

302312
// GetStatus implements the baseserver.Service interface.
@@ -335,6 +345,7 @@ func (kc *KTranslate) HttpInfo(w http.ResponseWriter, r *http.Request) {
335345
Sinks: map[ss.Sink]map[string]float64{},
336346
SnmpDeviceData: map[string]map[string]float64{},
337347
Inputs: map[string]map[string]float64{},
348+
Stitcher: kc.stitcher.HttpInfo(),
338349
}
339350

340351
// Now, let other sinks do their work

‎pkg/cat/types.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"github.com/kentik/ktranslate/pkg/maps"
2121
"github.com/kentik/ktranslate/pkg/rollup"
2222
"github.com/kentik/ktranslate/pkg/sinks"
23+
"github.com/kentik/ktranslate/pkg/stitch"
2324
"github.com/kentik/ktranslate/pkg/util/enrich"
2425
"github.com/kentik/ktranslate/pkg/util/gopatricia/patricia"
2526
"github.com/kentik/ktranslate/pkg/util/resolv"
@@ -86,6 +87,7 @@ type KTranslate struct {
8687
shutdown func(string)
8788
objmgr sinks.CloudObjectManager
8889
tee sinks.SinkImpl
90+
stitcher *stitch.Stitcher
8991
}
9092

9193
type CustomMapper struct {
@@ -117,6 +119,7 @@ type hc struct {
117119
Sinks map[sinks.Sink]map[string]float64
118120
SnmpDeviceData map[string]map[string]float64
119121
Inputs map[string]map[string]float64
122+
Stitcher map[string]float64
120123
}
121124

122125
type Flow struct {

‎pkg/kt/testing.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,4 +16,9 @@ var (
1616
&JCHF{CompanyId: 10, SrcAddr: "10.2.2.1", Protocol: "UDP", DstAddr: "2001:db8::68", Timestamp: 1, L4DstPort: 80, OutputPort: IfaceID(20), EventType: KENTIK_EVENT_SNMP_INT_METRIC, CustomStr: map[string]string{"foo": "bar"}, CustomInt: map[string]int32{"foo": 1}, CustomBigInt: map[string]int64{"foo": 12}, InBytes: 12121, InPkts: 12, OutBytes: 13, OutPkts: 1, SrcEthMac: "90:61:ae:fb:c2:19", avroSet: map[string]interface{}{}},
1717
&JCHF{CompanyId: 10, SrcAddr: "10.2.2.1", Protocol: "UDP", DstAddr: "2001:db8::68", Timestamp: 1, L4DstPort: 90, OutputPort: IfaceID(40), EventType: KENTIK_EVENT_SNMP_INT_METRIC, CustomStr: map[string]string{"foo": "sas"}, CustomInt: map[string]int32{"foo": 2}, CustomBigInt: map[string]int64{"foo": 22}, InBytes: 12121, InPkts: 12, OutBytes: 13, OutPkts: 1, SrcEthMac: "90:61:ae:fb:c2:19", avroSet: map[string]interface{}{}},
1818
}
19+
20+
InputTestingUnify = []*JCHF{
21+
&JCHF{CompanyId: 10, SrcAddr: "10.2.2.1", Protocol: "TCP", DstAddr: "2001:db8::68", Timestamp: 1, L4DstPort: 80, L4SrcPort: 34566, SrcAs: 1111, SrcGeo: "US", OutputPort: IfaceID(20), EventType: KENTIK_EVENT_TYPE, CustomStr: map[string]string{"foo": "bar"}, CustomInt: map[string]int32{"fooI": 1}, CustomBigInt: map[string]int64{"fooII": 12}, InBytes: 12121, InPkts: 12, OutBytes: 13, OutPkts: 1, SrcEthMac: "90:61:ae:fb:c2:19", avroSet: map[string]interface{}{}},
22+
&JCHF{CompanyId: 10, Protocol: "TCP", DstAddr: "10.2.2.1", SrcAddr: "2001:db8::68", Timestamp: 1, L4SrcPort: 80, L4DstPort: 34566, SrcAs: 1111, SrcGeo: "US", OutputPort: IfaceID(20), EventType: KENTIK_EVENT_TYPE, CustomStr: map[string]string{"foo": "bar"}, CustomInt: map[string]int32{"fooI": 1}, CustomBigInt: map[string]int64{"fooII": 12}, InBytes: 12121, InPkts: 12, OutBytes: 13, OutPkts: 1, SrcEthMac: "90:61:ae:fb:c2:19", avroSet: map[string]interface{}{}},
23+
}
1924
)

‎pkg/kt/types.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package kt
33
import (
44
"regexp"
55
"strconv"
6+
"strings"
67
"syscall"
78
"time"
89
)
@@ -335,6 +336,14 @@ func (j *JCHF) SetIFPorts(p IfaceID) *JCHF {
335336
return j
336337
}
337338

339+
func (j *JCHF) GetKey() string {
340+
if j.L4SrcPort < 32768 { // Guess if this is a ingress or egress flow.
341+
return strings.Join([]string{j.CustomStr["src_endpoint"], j.CustomStr["dst_endpoint"], j.Protocol}, "|")
342+
} else {
343+
return strings.Join([]string{j.CustomStr["dst_endpoint"], j.CustomStr["src_endpoint"], j.Protocol}, "|")
344+
}
345+
}
346+
338347
type AgentId IntId
339348

340349
func NewAgentId(id string) AgentId {

0 commit comments

Comments
 (0)