feat: expose nexthop as dimension

common/schema/definition.go

@@ -94,6 +94,7 @@ const (
 	ColumnICMPv6
 	ColumnICMPv6Type
 	ColumnICMPv6Code
+	ColumnNextHop
 
 	ColumnLast
 )
@@ -366,7 +367,8 @@ END`,
 				ClickHouseAlias: `if(Proto = 1, ` +
 					`dictGetOrDefault('icmp', 'name', tuple(Proto, ICMPv4Type, ICMPv4Code), ` +
 					`concat(toString(ICMPv4Type), '/', toString(ICMPv4Code))), '')`,
-			}, {
+			},
+			{
 				Key:            ColumnICMPv6,
 				Depends:        []ColumnKey{ColumnProto, ColumnICMPv6Type, ColumnICMPv6Code},
 				Disabled:       true,
@@ -376,6 +378,11 @@ END`,
 					`dictGetOrDefault('icmp', 'name', tuple(Proto, ICMPv6Type, ICMPv6Code), ` +
 					`concat(toString(ICMPv6Type), '/', toString(ICMPv6Code))), '')`,
 			},
+			{
+				Key:             ColumnNextHop,
+				ClickHouseType:  "IPv6",
+				ClickHouseCodec: "ZSTD(1)",
+			},
 		},
 	}.finalize()
 }

common/schema/protobuf.go

@@ -106,6 +106,7 @@ func (schema *Schema) ProtobufMarshal(bf *FlowMessage) []byte {
 	schema.ProtobufAppendVarint(bf, ColumnDstNetMask, uint64(bf.DstNetMask))
 	schema.ProtobufAppendIP(bf, ColumnSrcAddr, bf.SrcAddr)
 	schema.ProtobufAppendIP(bf, ColumnDstAddr, bf.DstAddr)
+	schema.ProtobufAppendIP(bf, ColumnNextHop, bf.NextHop)
 	if !schema.IsDisabled(ColumnGroupL2) {
 		schema.ProtobufAppendVarint(bf, ColumnSrcVlan, uint64(bf.SrcVlan))
 		schema.ProtobufAppendVarint(bf, ColumnDstVlan, uint64(bf.DstVlan))

console/config_test.go

@@ -79,6 +79,7 @@ func TestConfigHandler(t *testing.T) {
 					"DstPort",
 					"PacketSizeBucket",
 					"ForwardingStatus",
+					"NextHop",
 				},
 				"truncatable": []string{"SrcAddr", "DstAddr"},
 			},

console/data/docs/02-configuration.md

@@ -254,7 +254,7 @@ The following configuration keys are accepted:
   from flow except if the ASN is private), `geoip`, `routing`, and
   `routing-except-private`. The default value is `flow`, `routing`, and
   `geoip`.
-- `net-providers` defines the sources for prefix lengths. `flow` uses the value
+- `net-providers` defines the sources for prefix lengths and nexthop. `flow` uses the value
   provided by the flow message (if any), while `routing` looks it up using the BMP
   component. If multiple sources are provided, the value of the first source
   providing a non-default route is taken. The default value is `flow` and `routing`.

console/filter/parser.peg

@@ -54,6 +54,7 @@ ColumnIP ←
  / "DstAddr"i !IdentStart #{ return c.metaColumn("DstAddr") } { return c.acceptColumn() }
  / "SrcAddrNAT"i !IdentStart #{ return c.metaColumn("SrcAddrNAT") } { return c.acceptColumn() }
  / "DstAddrNAT"i !IdentStart #{ return c.metaColumn("DstAddrNAT") } { return c.acceptColumn() }
+ / "NextHop"i !IdentStart #{ return c.metaColumn("NextHop") } { return c.acceptColumn() }
 ConditionIPExpr "condition on IP" ←
    column:ColumnIP _
    operator:("=" / "!=") _ ip:IP {

inlet/core/enricher.go

@@ -103,6 +103,9 @@ func (c *Component) enrichFlow(exporterIP netip.Addr, exporterStr string, flow *
 	flow.SrcNetMask = c.getNetMask(flow.SrcNetMask, sourceRouting.NetMask)
 	flow.DstNetMask = c.getNetMask(flow.DstNetMask, destRouting.NetMask)
 
+	// set next hop according to user config
+	flow.NextHop = c.getNextHop(flow.NextHop, destRouting.NextHop)
+
 	// set asns according to user config
 	flow.SrcAS = c.getASNumber(flow.SrcAddr, flow.SrcAS, sourceRouting.ASN)
 	flow.DstAS = c.getASNumber(flow.DstAddr, flow.DstAS, destRouting.ASN)
@@ -176,6 +179,22 @@ func (c *Component) getNetMask(flowMask, bmpMask uint8) (mask uint8) {
 	return mask
 }
 
+func (c *Component) getNextHop(flowNextHop netip.Addr, bmpNextHop netip.Addr) (nextHop netip.Addr) {
+	nextHop = netip.IPv6Unspecified()
+	for _, provider := range c.config.NetProviders {
+		if !nextHop.IsUnspecified() {
+			break
+		}
+		switch provider {
+		case NetProviderFlow:
+			nextHop = flowNextHop
+		case NetProviderRouting:
+			nextHop = bmpNextHop
+		}
+	}
+	return nextHop
+}
+
 func (c *Component) writeExporter(flow *schema.FlowMessage, classification exporterClassification) bool {
 	if classification.Reject {
 		return false

inlet/core/enricher_test.go

@@ -693,3 +693,61 @@ func TestGetNetMask(t *testing.T) {
 		})
 	}
 }
+
+func TestGetNextHop(t *testing.T) {
+	nh1 := netip.MustParseAddr("2001:db8::1")
+	nh2 := netip.MustParseAddr("2001:db8::2")
+	cases := []struct {
+		FlowNextHop    netip.Addr
+		RoutingNextHop netip.Addr
+		Providers      []NetProvider
+		Expected       netip.Addr
+	}{
+		// Flow
+		{netip.IPv6Unspecified(), netip.IPv6Unspecified(), []NetProvider{NetProviderFlow}, netip.IPv6Unspecified()},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderFlow}, nh1},
+		{netip.IPv6Unspecified(), nh1, []NetProvider{NetProviderFlow}, netip.IPv6Unspecified()},
+		// Routing
+		{netip.IPv6Unspecified(), netip.IPv6Unspecified(), []NetProvider{NetProviderRouting}, netip.IPv6Unspecified()},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderRouting}, netip.IPv6Unspecified()},
+		{netip.IPv6Unspecified(), nh1, []NetProvider{NetProviderRouting}, nh1},
+		// Both
+		{netip.IPv6Unspecified(), netip.IPv6Unspecified(), []NetProvider{NetProviderRouting, NetProviderFlow}, netip.IPv6Unspecified()},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderRouting, NetProviderFlow}, nh1},
+		{netip.IPv6Unspecified(), nh2, []NetProvider{NetProviderRouting, NetProviderFlow}, nh2},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderRouting, NetProviderFlow}, nh1},
+		{nh1, nh2, []NetProvider{NetProviderRouting, NetProviderFlow}, nh2},
+
+		{netip.IPv6Unspecified(), netip.IPv6Unspecified(), []NetProvider{NetProviderFlow, NetProviderRouting}, netip.IPv6Unspecified()},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderFlow, NetProviderRouting}, nh1},
+		{netip.IPv6Unspecified(), nh2, []NetProvider{NetProviderFlow, NetProviderRouting}, nh2},
+		{nh1, netip.IPv6Unspecified(), []NetProvider{NetProviderFlow, NetProviderRouting}, nh1},
+		{nh1, nh2, []NetProvider{NetProviderFlow, NetProviderRouting}, nh1},
+	}
+	for i, tc := range cases {
+		i++
+		t.Run(fmt.Sprintf("case %d", i), func(t *testing.T) {
+			r := reporter.NewMock(t)
+
+			// We don't need all components as we won't start the component.
+			configuration := DefaultConfiguration()
+			configuration.NetProviders = tc.Providers
+			routingComponent := routing.NewMock(t, r)
+			routingComponent.PopulateRIB(t)
+
+			c, err := New(r, configuration, Dependencies{
+				Daemon:  daemon.NewMock(t),
+				GeoIP:   geoip.NewMock(t, r),
+				Routing: routingComponent,
+				Schema:  schema.NewMock(t),
+			})
+			if err != nil {
+				t.Fatalf("New() error:\n%+v", err)
+			}
+			got := c.getNextHop(tc.FlowNextHop, tc.RoutingNextHop)
+			if diff := helpers.Diff(got, tc.Expected); diff != "" {
+				t.Fatalf("getNextHop() (-got, +want):\n%s", diff)
+			}
+		})
+	}
+}

inlet/routing/provider/bioris/root.go

@@ -294,6 +294,16 @@ func (p *Provider) lpmResponseToLookupResult(lpm *pb.LPMResponse) (bmp.LookupRes
 	}
 
 	res.NetMask = uint8(r.Pfx.GetLength())
+	nh := pfx.BgpPath.GetNextHop()
+	if nh != nil {
+		bnh := bnet.IPFromProtoIP(nh)
+		nhAddr, ok := netip.AddrFromSlice(bnh.ToNetIP())
+		if !ok {
+			return res, fmt.Errorf("lpm: invalid next hop")
+		}
+		res.NextHop = nhAddr
+	}
+
 	return res, nil
 }
 

inlet/routing/provider/bioris/root_test.go

@@ -451,6 +451,7 @@ func TestBioRIS(t *testing.T) {
 					LocalData2: 4,
 				},
 			},
+			NextHop: netip.MustParseAddr("2001:db8::7"),
 		}
 		if diff := helpers.Diff(got, expected); diff != "" {
 			t.Errorf("Lookup() (-got, +want):\n%s", diff)

inlet/routing/provider/bmp/lookup.go

@@ -53,7 +53,11 @@ func (p *Provider) Lookup(_ context.Context, ip netip.Addr, nh netip.Addr, _ net
 	if len(routes) == 0 {
 		return LookupResult{}, fmt.Errorf("no route found for %s", ip)
 	}
-	attributes := p.rib.rtas.Get(routes[len(routes)-1].attributes)
+	route := routes[len(routes)-1]
+	attributes := p.rib.rtas.Get(route.attributes)
+	// the next hop is updated from the rib in every case, because the user "opted in" for bmp as source if the lookup result is evaluated
+	nh = netip.Addr(p.rib.nextHops.Get(route.nextHop))
+
 	// prefix len is v6 coded in the bmp rib. We need to substract 96 if it's a v4 prefix
 	plen := attributes.plen
 	if ip.Is4() || ip.Is4In6() {
@@ -65,5 +69,6 @@ func (p *Provider) Lookup(_ context.Context, ip netip.Addr, nh netip.Addr, _ net
 		Communities:      attributes.communities,
 		LargeCommunities: attributes.largeCommunities,
 		NetMask:          plen,
+		NextHop:          nh,
 	}, nil
 }

inlet/routing/provider/root.go

@@ -23,6 +23,7 @@ type LookupResult struct {
 	Communities      []uint32
 	LargeCommunities []bgp.LargeCommunity
 	NetMask          uint8
+	NextHop          netip.Addr
 }
 
 // Dependencies are the dependencies for a provider.

orchestrator/clickhouse/testdata/states/012.csv

@@ -0,0 +1,15 @@
+"asns","CREATE DICTIONARY default.asns (`asn` UInt32 INJECTIVE, `name` String) PRIMARY KEY asn SOURCE(HTTP(URL 'http://something/api/v0/orchestrator/clickhouse/asns.csv' FORMAT 'CSVWithNames')) LIFETIME(MIN 0 MAX 3600) LAYOUT(HASHED()) SETTINGS(format_csv_allow_single_quotes = 0)"
+"icmp","CREATE DICTIONARY default.icmp (`proto` UInt8, `type` UInt8, `code` UInt8, `name` String) PRIMARY KEY proto, type, code SOURCE(HTTP(URL 'http://something/api/v0/orchestrator/clickhouse/icmp.csv' FORMAT 'CSVWithNames')) LIFETIME(MIN 0 MAX 3600) LAYOUT(COMPLEX_KEY_HASHED()) SETTINGS(format_csv_allow_single_quotes = 0)"
+"flows","CREATE TABLE default.flows (`TimeReceived` DateTime CODEC(DoubleDelta, LZ4), `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAddr` IPv6 CODEC(ZSTD(1)), `DstAddr` IPv6 CODEC(ZSTD(1)), `SrcNetMask` UInt8, `DstNetMask` UInt8, `SrcNetPrefix` String ALIAS multiIf(EType = 2048, concat(replaceRegexpOne(CAST(IPv6CIDRToRange(SrcAddr, CAST(96 + SrcNetMask, 'UInt8')).1, 'String'), '^::ffff:', ''), '/', CAST(SrcNetMask, 'String')), EType = 34525, concat(CAST(IPv6CIDRToRange(SrcAddr, SrcNetMask).1, 'String'), '/', CAST(SrcNetMask, 'String')), ''), `DstNetPrefix` String ALIAS multiIf(EType = 2048, concat(replaceRegexpOne(CAST(IPv6CIDRToRange(DstAddr, CAST(96 + DstNetMask, 'UInt8')).1, 'String'), '^::ffff:', ''), '/', CAST(DstNetMask, 'String')), EType = 34525, concat(CAST(IPv6CIDRToRange(DstAddr, DstNetMask).1, 'String'), '/', CAST(DstNetMask, 'String')), ''), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `DstASPath` Array(UInt32), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `DstCommunities` Array(UInt32), `DstLargeCommunities` Array(UInt128), `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `SrcPort` UInt16, `DstPort` UInt16, `Bytes` UInt64 CODEC(T64, LZ4), `Packets` UInt64 CODEC(T64, LZ4), `PacketSize` UInt64 ALIAS intDiv(Bytes, Packets), `PacketSizeBucket` LowCardinality(String) ALIAS multiIf(PacketSize < 64, '0-63', PacketSize < 128, '64-127', PacketSize < 256, '128-255', PacketSize < 512, '256-511', PacketSize < 768, '512-767', PacketSize < 1024, '768-1023', PacketSize < 1280, '1024-1279', PacketSize < 1501, '1280-1500', PacketSize < 2048, '1501-2047', PacketSize < 3072, '2048-3071', PacketSize < 4096, '3072-4095', PacketSize < 8192, '4096-8191', PacketSize < 10240, '8192-10239', PacketSize < 16384, '10240-16383', PacketSize < 32768, '16384-32767', PacketSize < 65536, '32768-65535', '65536-Inf'), `ForwardingStatus` UInt32, `NextHop` IPv6 CODEC(ZSTD(1))) ENGINE = MergeTree PARTITION BY toYYYYMMDDhhmmss(toStartOfInterval(TimeReceived, toIntervalSecond(25920))) ORDER BY (TimeReceived, ExporterAddress, InIfName, OutIfName) TTL TimeReceived + toIntervalSecond(1296000) SETTINGS index_granularity = 8192"
+"networks","CREATE DICTIONARY default.networks (`network` String, `name` String, `role` String, `site` String, `region` String, `tenant` String) PRIMARY KEY network SOURCE(HTTP(URL 'http://something/api/v0/orchestrator/clickhouse/networks.csv' FORMAT 'CSVWithNames')) LIFETIME(MIN 0 MAX 3600) LAYOUT(IP_TRIE()) SETTINGS(format_csv_allow_single_quotes = 0)"
+"exporters","CREATE MATERIALIZED VIEW default.exporters (`TimeReceived` DateTime, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `IfName` String, `IfDescription` String, `IfSpeed` UInt32, `IfConnectivity` String, `IfProvider` String, `IfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2)) ENGINE = ReplacingMergeTree(TimeReceived) ORDER BY (ExporterAddress, IfName) TTL TimeReceived + toIntervalDay(1) SETTINGS index_granularity = 8192 AS SELECT DISTINCT TimeReceived, ExporterAddress, ExporterName, ExporterGroup, ExporterRole, ExporterSite, ExporterRegion, ExporterTenant, [InIfName, OutIfName][num] AS IfName, [InIfDescription, OutIfDescription][num] AS IfDescription, [InIfSpeed, OutIfSpeed][num] AS IfSpeed, [InIfConnectivity, OutIfConnectivity][num] AS IfConnectivity, [InIfProvider, OutIfProvider][num] AS IfProvider, [InIfBoundary, OutIfBoundary][num] AS IfBoundary FROM default.flows ARRAY JOIN arrayEnumerate([1, 2]) AS num"
+"protocols","CREATE DICTIONARY default.protocols (`proto` UInt8 INJECTIVE, `name` String, `description` String) PRIMARY KEY proto SOURCE(HTTP(URL 'http://something/api/v0/orchestrator/clickhouse/protocols.csv' FORMAT 'CSVWithNames')) LIFETIME(MIN 0 MAX 3600) LAYOUT(HASHED()) SETTINGS(format_csv_allow_single_quotes = 0)"
+"flows_1m0s","CREATE TABLE default.flows_1m0s (`TimeReceived` DateTime CODEC(DoubleDelta, LZ4), `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64 CODEC(T64, LZ4), `Packets` UInt64 CODEC(T64, LZ4), `PacketSize` UInt64 ALIAS intDiv(Bytes, Packets), `PacketSizeBucket` LowCardinality(String) ALIAS multiIf(PacketSize < 64, '0-63', PacketSize < 128, '64-127', PacketSize < 256, '128-255', PacketSize < 512, '256-511', PacketSize < 768, '512-767', PacketSize < 1024, '768-1023', PacketSize < 1280, '1024-1279', PacketSize < 1501, '1280-1500', PacketSize < 2048, '1501-2047', PacketSize < 3072, '2048-3071', PacketSize < 4096, '3072-4095', PacketSize < 8192, '4096-8191', PacketSize < 10240, '8192-10239', PacketSize < 16384, '10240-16383', PacketSize < 32768, '16384-32767', PacketSize < 65536, '32768-65535', '65536-Inf'), `ForwardingStatus` UInt32, `NextHop` IPv6 CODEC(ZSTD(1))) ENGINE = SummingMergeTree((Bytes, Packets)) PARTITION BY toYYYYMMDDhhmmss(toStartOfInterval(TimeReceived, toIntervalSecond(12096))) PRIMARY KEY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate) ORDER BY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, NextHop) TTL TimeReceived + toIntervalSecond(604800) SETTINGS index_granularity = 8192"
+"flows_5m0s","CREATE TABLE default.flows_5m0s (`TimeReceived` DateTime CODEC(DoubleDelta, LZ4), `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64 CODEC(T64, LZ4), `Packets` UInt64 CODEC(T64, LZ4), `PacketSize` UInt64 ALIAS intDiv(Bytes, Packets), `PacketSizeBucket` LowCardinality(String) ALIAS multiIf(PacketSize < 64, '0-63', PacketSize < 128, '64-127', PacketSize < 256, '128-255', PacketSize < 512, '256-511', PacketSize < 768, '512-767', PacketSize < 1024, '768-1023', PacketSize < 1280, '1024-1279', PacketSize < 1501, '1280-1500', PacketSize < 2048, '1501-2047', PacketSize < 3072, '2048-3071', PacketSize < 4096, '3072-4095', PacketSize < 8192, '4096-8191', PacketSize < 10240, '8192-10239', PacketSize < 16384, '10240-16383', PacketSize < 32768, '16384-32767', PacketSize < 65536, '32768-65535', '65536-Inf'), `ForwardingStatus` UInt32, `NextHop` IPv6 CODEC(ZSTD(1))) ENGINE = SummingMergeTree((Bytes, Packets)) PARTITION BY toYYYYMMDDhhmmss(toStartOfInterval(TimeReceived, toIntervalSecond(155520))) PRIMARY KEY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate) ORDER BY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, NextHop) TTL TimeReceived + toIntervalSecond(7776000) SETTINGS index_granularity = 8192"
+"flows_1h0m0s","CREATE TABLE default.flows_1h0m0s (`TimeReceived` DateTime CODEC(DoubleDelta, LZ4), `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64 CODEC(T64, LZ4), `Packets` UInt64 CODEC(T64, LZ4), `PacketSize` UInt64 ALIAS intDiv(Bytes, Packets), `PacketSizeBucket` LowCardinality(String) ALIAS multiIf(PacketSize < 64, '0-63', PacketSize < 128, '64-127', PacketSize < 256, '128-255', PacketSize < 512, '256-511', PacketSize < 768, '512-767', PacketSize < 1024, '768-1023', PacketSize < 1280, '1024-1279', PacketSize < 1501, '1280-1500', PacketSize < 2048, '1501-2047', PacketSize < 3072, '2048-3071', PacketSize < 4096, '3072-4095', PacketSize < 8192, '4096-8191', PacketSize < 10240, '8192-10239', PacketSize < 16384, '10240-16383', PacketSize < 32768, '16384-32767', PacketSize < 65536, '32768-65535', '65536-Inf'), `ForwardingStatus` UInt32, `NextHop` IPv6 CODEC(ZSTD(1))) ENGINE = SummingMergeTree((Bytes, Packets)) PARTITION BY toYYYYMMDDhhmmss(toStartOfInterval(TimeReceived, toIntervalSecond(622080))) PRIMARY KEY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate) ORDER BY (TimeReceived, ExporterAddress, EType, Proto, InIfName, SrcAS, ForwardingStatus, OutIfName, DstAS, SamplingRate, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, NextHop) TTL TimeReceived + toIntervalSecond(31104000) SETTINGS index_granularity = 8192"
+"flows_1m0s_consumer","CREATE MATERIALIZED VIEW default.flows_1m0s_consumer TO default.flows_1m0s (`TimeReceived` DateTime, `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64, `Packets` UInt64, `ForwardingStatus` UInt32, `NextHop` IPv6) AS SELECT toStartOfInterval(TimeReceived, toIntervalSecond(60)) AS TimeReceived, SamplingRate, ExporterAddress, ExporterName, ExporterGroup, ExporterRole, ExporterSite, ExporterRegion, ExporterTenant, SrcAS, DstAS, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, InIfName, OutIfName, InIfDescription, OutIfDescription, InIfSpeed, OutIfSpeed, InIfConnectivity, OutIfConnectivity, InIfProvider, OutIfProvider, InIfBoundary, OutIfBoundary, EType, Proto, Bytes, Packets, ForwardingStatus, NextHop FROM default.flows"
+"flows_5m0s_consumer","CREATE MATERIALIZED VIEW default.flows_5m0s_consumer TO default.flows_5m0s (`TimeReceived` DateTime, `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64, `Packets` UInt64, `ForwardingStatus` UInt32, `NextHop` IPv6) AS SELECT toStartOfInterval(TimeReceived, toIntervalSecond(300)) AS TimeReceived, SamplingRate, ExporterAddress, ExporterName, ExporterGroup, ExporterRole, ExporterSite, ExporterRegion, ExporterTenant, SrcAS, DstAS, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, InIfName, OutIfName, InIfDescription, OutIfDescription, InIfSpeed, OutIfSpeed, InIfConnectivity, OutIfConnectivity, InIfProvider, OutIfProvider, InIfBoundary, OutIfBoundary, EType, Proto, Bytes, Packets, ForwardingStatus, NextHop FROM default.flows"
+"flows_1h0m0s_consumer","CREATE MATERIALIZED VIEW default.flows_1h0m0s_consumer TO default.flows_1h0m0s (`TimeReceived` DateTime, `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` LowCardinality(String), `DstNetName` LowCardinality(String), `SrcNetRole` LowCardinality(String), `DstNetRole` LowCardinality(String), `SrcNetSite` LowCardinality(String), `DstNetSite` LowCardinality(String), `SrcNetRegion` LowCardinality(String), `DstNetRegion` LowCardinality(String), `SrcNetTenant` LowCardinality(String), `DstNetTenant` LowCardinality(String), `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `Bytes` UInt64, `Packets` UInt64, `ForwardingStatus` UInt32, `NextHop` IPv6) AS SELECT toStartOfInterval(TimeReceived, toIntervalSecond(3600)) AS TimeReceived, SamplingRate, ExporterAddress, ExporterName, ExporterGroup, ExporterRole, ExporterSite, ExporterRegion, ExporterTenant, SrcAS, DstAS, SrcNetName, DstNetName, SrcNetRole, DstNetRole, SrcNetSite, DstNetSite, SrcNetRegion, DstNetRegion, SrcNetTenant, DstNetTenant, SrcCountry, DstCountry, Dst1stAS, Dst2ndAS, Dst3rdAS, InIfName, OutIfName, InIfDescription, OutIfDescription, InIfSpeed, OutIfSpeed, InIfConnectivity, OutIfConnectivity, InIfProvider, OutIfProvider, InIfBoundary, OutIfBoundary, EType, Proto, Bytes, Packets, ForwardingStatus, NextHop FROM default.flows"
+"flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw","CREATE TABLE default.flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw (`TimeReceived` DateTime CODEC(DoubleDelta, LZ4), `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAddr` IPv6 CODEC(ZSTD(1)), `DstAddr` IPv6 CODEC(ZSTD(1)), `SrcNetMask` UInt8, `DstNetMask` UInt8, `SrcAS` UInt32, `DstAS` UInt32, `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `DstASPath` Array(UInt32), `DstCommunities` Array(UInt32), `DstLargeCommunitiesASN` Array(UInt32), `DstLargeCommunitiesLocalData1` Array(UInt32), `DstLargeCommunitiesLocalData2` Array(UInt32), `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `SrcPort` UInt16, `DstPort` UInt16, `Bytes` UInt64 CODEC(T64, LZ4), `Packets` UInt64 CODEC(T64, LZ4), `ForwardingStatus` UInt32, `NextHop` IPv6 CODEC(ZSTD(1))) ENGINE = Kafka SETTINGS kafka_broker_list = '127.0.0.1:9092', kafka_topic_list = 'flows-UHC7Y5U3MEDTTEOKH6IF3NGWXQ', kafka_group_name = 'clickhouse', kafka_format = 'Protobuf', kafka_schema = 'flow-UHC7Y5U3MEDTTEOKH6IF3NGWXQ.proto:FlowMessagevUHC7Y5U3MEDTTEOKH6IF3NGWXQ', kafka_num_consumers = 1, kafka_thread_per_consumer = 1, kafka_handle_error_mode = 'stream'"
+"flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw_errors","CREATE MATERIALIZED VIEW default.flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw_errors (`timestamp` DateTime, `topic` LowCardinality(String), `partition` UInt64, `offset` UInt64, `raw` String, `error` String) ENGINE = MergeTree PARTITION BY toYYYYMMDDhhmmss(toStartOfHour(timestamp)) ORDER BY (timestamp, topic, partition, offset) TTL timestamp + toIntervalDay(1) SETTINGS index_granularity = 8192 AS SELECT now() AS timestamp, _topic AS topic, _partition AS partition, _offset AS offset, _raw_message AS raw, _error AS error FROM default.flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw WHERE length(_error) > 0"
+"flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw_consumer","CREATE MATERIALIZED VIEW default.flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw_consumer TO default.flows (`TimeReceived` DateTime, `SamplingRate` UInt64, `ExporterAddress` LowCardinality(IPv6), `ExporterName` LowCardinality(String), `ExporterGroup` LowCardinality(String), `ExporterRole` LowCardinality(String), `ExporterSite` LowCardinality(String), `ExporterRegion` LowCardinality(String), `ExporterTenant` LowCardinality(String), `SrcAddr` IPv6, `DstAddr` IPv6, `SrcNetMask` UInt8, `DstNetMask` UInt8, `SrcAS` UInt32, `DstAS` UInt32, `SrcNetName` String, `DstNetName` String, `SrcNetRole` String, `DstNetRole` String, `SrcNetSite` String, `DstNetSite` String, `SrcNetRegion` String, `DstNetRegion` String, `SrcNetTenant` String, `DstNetTenant` String, `SrcCountry` FixedString(2), `DstCountry` FixedString(2), `DstASPath` Array(UInt32), `Dst1stAS` UInt32, `Dst2ndAS` UInt32, `Dst3rdAS` UInt32, `DstCommunities` Array(UInt32), `DstLargeCommunities` Array(UInt128), `InIfName` LowCardinality(String), `OutIfName` LowCardinality(String), `InIfDescription` LowCardinality(String), `OutIfDescription` LowCardinality(String), `InIfSpeed` UInt32, `OutIfSpeed` UInt32, `InIfConnectivity` LowCardinality(String), `OutIfConnectivity` LowCardinality(String), `InIfProvider` LowCardinality(String), `OutIfProvider` LowCardinality(String), `InIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `OutIfBoundary` Enum8('undefined' = 0, 'external' = 1, 'internal' = 2), `EType` UInt32, `Proto` UInt32, `SrcPort` UInt16, `DstPort` UInt16, `Bytes` UInt64, `Packets` UInt64, `ForwardingStatus` UInt32, `NextHop` IPv6) AS WITH arrayCompact(DstASPath) AS c_DstASPath SELECT TimeReceived, SamplingRate, ExporterAddress, ExporterName, ExporterGroup, ExporterRole, ExporterSite, ExporterRegion, ExporterTenant, SrcAddr, DstAddr, SrcNetMask, DstNetMask, SrcAS, DstAS, dictGetOrDefault('default.networks', 'name', SrcAddr, '') AS SrcNetName, dictGetOrDefault('default.networks', 'name', DstAddr, '') AS DstNetName, dictGetOrDefault('default.networks', 'role', SrcAddr, '') AS SrcNetRole, dictGetOrDefault('default.networks', 'role', DstAddr, '') AS DstNetRole, dictGetOrDefault('default.networks', 'site', SrcAddr, '') AS SrcNetSite, dictGetOrDefault('default.networks', 'site', DstAddr, '') AS DstNetSite, dictGetOrDefault('default.networks', 'region', SrcAddr, '') AS SrcNetRegion, dictGetOrDefault('default.networks', 'region', DstAddr, '') AS DstNetRegion, dictGetOrDefault('default.networks', 'tenant', SrcAddr, '') AS SrcNetTenant, dictGetOrDefault('default.networks', 'tenant', DstAddr, '') AS DstNetTenant, SrcCountry, DstCountry, DstASPath, c_DstASPath[1] AS Dst1stAS, c_DstASPath[2] AS Dst2ndAS, c_DstASPath[3] AS Dst3rdAS, DstCommunities, arrayMap((asn, l1, l2) -> ((bitShiftLeft(CAST(asn, 'UInt128'), 64) + bitShiftLeft(CAST(l1, 'UInt128'), 32)) + CAST(l2, 'UInt128')), DstLargeCommunitiesASN, DstLargeCommunitiesLocalData1, DstLargeCommunitiesLocalData2) AS DstLargeCommunities, InIfName, OutIfName, InIfDescription, OutIfDescription, InIfSpeed, OutIfSpeed, InIfConnectivity, OutIfConnectivity, InIfProvider, OutIfProvider, InIfBoundary, OutIfBoundary, EType, Proto, SrcPort, DstPort, Bytes, Packets, ForwardingStatus, NextHop FROM default.flows_UHC7Y5U3MEDTTEOKH6IF3NGWXQ_raw WHERE length(_error) = 0"