Support reporting connection count for incoming Internet connections (#297)

Author: zohar7ch

src/mapper/pkg/cloudclient/generated.go

@@ -7,14 +7,17 @@ directive @constraint(
 	example: String!
 ) on ENUM_VALUE
 
-"""The @defer directive may be specified on a fragment spread to imply de-prioritization, that causes the fragment to be omitted in the initial response, and delivered as a subsequent response afterward. A query with @defer directive will cause the request to potentially return multiple responses, where non-deferred data is delivered in the initial response and data deferred delivered in a subsequent response. @include and @skip take precedence over @defer."""
+"""Directs the executor to defer this fragment when the `if` argument is true or undefined."""
 directive @defer(
+"""Deferred when true or undefined."""
 	if: Boolean
+"""Unique name"""
 	label: String
 ) on FRAGMENT_SPREAD | INLINE_FRAGMENT
 
-"""The @deprecated built-in directive is used within the type system definition language to indicate deprecated portions of a GraphQL service's schema, such as deprecated fields on a type, arguments on a field, input fields on an input type, or values of an enum type."""
+"""Marks an element of a GraphQL schema as no longer supported."""
 directive @deprecated(
+"""Explains why this element was deprecated, usually also including a suggestion for how to access supported similar data. Formatted using the Markdown syntax, as specified by [CommonMark](https://commonmark.org/)."""
 	reason: String
 ) on FIELD_DEFINITION | ARGUMENT_DEFINITION | INPUT_FIELD_DEFINITION | ENUM_VALUE
 
@@ -27,8 +30,9 @@ directive @httpError(
 	statusCode: Int!
 ) on ENUM_VALUE
 
-"""The @include directive may be provided for fields, fragment spreads, and inline fragments, and allows for conditional inclusion during execution as described by the if argument."""
+"""Directs the executor to include this field or fragment only when the `if` argument is true."""
 directive @include(
+"""Included when true."""
 	if: Boolean!
 ) on FIELD | FRAGMENT_SPREAD | INLINE_FRAGMENT
 
@@ -41,6 +45,9 @@ user authentication, meaning anyone and everyone can execute it. USE WITH CAUTIO
 user authentication, meaning anyone and everyone can execute it. USE WITH CAUTION."""
 directive @noauth on FIELD_DEFINITION
 
+"""Indicates exactly one field must be supplied and this field must not be `null`."""
+directive @oneOf on INPUT_OBJECT
+
 """@requiresRole indicates that the specified query / mutation / subscription requires any of the provided roles to be executed.
 Users without any of the specified roles will not be able to execute the query / mutation / subscription."""
 directive @requiresRole(
@@ -57,13 +64,15 @@ directive @restApiRoute(
 	tags: [String!]!
 ) on FIELD_DEFINITION
 
-"""The @skip directive may be provided for fields, fragment spreads, and inline fragments, and allows for conditional exclusion during execution as described by the if argument."""
+"""Directs the executor to skip this field or fragment when the `if` argument is true."""
 directive @skip(
+"""Skipped when true."""
 	if: Boolean!
 ) on FIELD | FRAGMENT_SPREAD | INLINE_FRAGMENT
 
-"""The @specifiedBy built-in directive is used within the type system definition language to provide a scalar specification URL for specifying the behavior of custom scalar types."""
+"""Exposes a URL that specifies the behavior of this scalar."""
 directive @specifiedBy(
+"""The URL that specifies the behavior of this scalar."""
 	url: String!
 ) on SCALAR
 
@@ -726,6 +735,8 @@ enum EdgeAccessStatusReason {
 	ALLOWED_BY_APPLIED_INTENTS_KAFKA_OVERLY_PERMISSIVE
 	ALLOWED_BY_APPLIED_INTENTS_DATABASE_OVERLY_PERMISSIVE
 	ALLOWED_BY_EXTERNAL_TRAFFIC_NETWORK_POLICY
+	ALLOWED_BY_INTERNET_EGRESS_NETWORK_POLICY
+	ALLOWED_BY_INTERNET_INGRESS_NETWORK_POLICY
 	WOULD_BE_ALLOWED_BY_EXTERNAL_TRAFFIC_NETWORK_POLICY
 	BLOCKED_BY_APPLIED_INTENTS_UNDER_PERMISSIVE
 	BLOCKED_BY_APPLIED_INTENTS_RESOURCE_MISMATCH
@@ -737,6 +748,8 @@ enum EdgeAccessStatusReason {
 	BLOCKED_BY_APPLIED_INTENTS_DATABASE_UNDER_PERMISSIVE
 	BLOCKED_BY_APPLIED_INTENTS_DATABASE_RESOURCE_MISMATCH
 	BLOCKED_BY_DATABASE_ENFORCEMENT_CONFIG_MISSING_APPLIED_INTENTS
+	BLOCKED_BY_INTERNET_EGRESS_NETWORK_POLICY
+	BLOCKED_BY_INTERNET_INGRESS_NETWORK_POLICY
 	BLOCKED_BY_DEFAULT_DENY
 	SHARED_SERVICE_ACCOUNT
 	CLIENT_ISTIO_SIDECAR_MISSING
@@ -1085,6 +1098,12 @@ enum IPFamily {
 	UNKNOWN
 }
 
+"""IP filters"""
+type IPFilterValue {
+	cidr: String!
+	exclude: [String!]
+}
+
 input IncomingInternetSourceInput {
 	ip: String!
 }
@@ -1098,6 +1117,7 @@ input IncomingTrafficIntentInput {
 	serverName: String!
 	namespace: String!
 	source: IncomingInternetSourceInput!
+	connectionsCount: ConnectionsCount
 }
 
 input IngressControllerConfigInput {
@@ -2288,6 +2308,7 @@ type NetworkPolicy {
 	workloadsAffected: Int!
 	spec: String!
 	lastUsed: Time
+	metadata: NetworkPolicyMetadata
 }
 
 input NetworkPolicyInput {
@@ -2302,6 +2323,12 @@ enum NetworkPolicyKind {
 	CILIUM_CLUSTER_WIDE_NETWORK_POLICY
 }
 
+type NetworkPolicyMetadata {
+	isEgress: Boolean!
+	isIngress: Boolean!
+	hasIpBlocks: Boolean!
+}
+
 enum NetworkPolicyScope {
 	PRIMARY
 	EGRESS

src/mapper/pkg/cloudclient/schema.graphql

@@ -184,6 +184,9 @@ func (c *CloudUploader) NotifyIncomingTrafficIntents(ctx context.Context, intent
 				},
 			},
 		}
+		if intent.ConnectionsCount != nil {
+			output.Intent.ConnectionsCount = nilable.FromPtr(intent.ConnectionsCount)
+		}
 		return output
 	})
 

src/mapper/pkg/clouduploader/cloud_upload.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"github.com/otterize/network-mapper/src/mapper/pkg/awsintentsholder"
 	"github.com/otterize/network-mapper/src/mapper/pkg/incomingtrafficholder"
+	"github.com/otterize/nilable"
 	"testing"
 	"time"
 
@@ -118,6 +119,7 @@ func (s *CloudUploaderTestSuite) TestUploadIncomingTrafficIntents() {
 		Server:   model.OtterizeServiceIdentity{Name: "server1", Namespace: s.testNamespace},
 		LastSeen: testTimestamp,
 		IP:       sourceIP,
+		SrcPorts: []int64{10},
 	}
 	s.incomingHolder.AddIntent(incomingIntent)
 
@@ -130,6 +132,7 @@ func (s *CloudUploaderTestSuite) TestUploadIncomingTrafficIntents() {
 				Source: cloudclient.IncomingInternetSourceInput{
 					Ip: sourceIP,
 				},
+				ConnectionsCount: nilable.From(cloudclient.ConnectionsCount{Current: lo.ToPtr(1), Added: lo.ToPtr(1), Removed: lo.ToPtr(0)}),
 			},
 		},
 	}

src/mapper/pkg/clouduploader/cloud_uploader_test.go

@@ -0,0 +1,73 @@
+package concurrentconnectioncounter
+
+import (
+	"github.com/otterize/network-mapper/src/mapper/pkg/cloudclient"
+	"github.com/samber/lo"
+)
+
+type ConnectionCounterMap[K comparable, Countable CountableIntent] map[K]*ConnectionCounter[Countable]
+
+type ConnectionCountDiffer[K comparable, Countable CountableIntent] struct {
+	currentCounters  ConnectionCounterMap[K, Countable]
+	previousCounters ConnectionCounterMap[K, Countable]
+}
+
+func NewConnectionCountDiffer[K comparable, Countable CountableIntent]() *ConnectionCountDiffer[K, Countable] {
+	return &ConnectionCountDiffer[K, Countable]{
+		currentCounters:  make(ConnectionCounterMap[K, Countable]),
+		previousCounters: make(ConnectionCounterMap[K, Countable]),
+	}
+}
+
+func (c *ConnectionCountDiffer[K, Countable]) Reset() {
+	c.previousCounters = c.currentCounters
+	c.currentCounters = make(ConnectionCounterMap[K, Countable])
+}
+
+func (c *ConnectionCountDiffer[K, Countable]) Increment(key K, counterInput CounterInput[Countable]) {
+
+	_, existingCounterFound := c.currentCounters[key]
+	if !existingCounterFound {
+		c.currentCounters[key] = NewConnectionCounter[Countable]()
+	}
+
+	c.currentCounters[key].AddConnection(counterInput)
+}
+
+func (c *ConnectionCountDiffer[K, Countable]) GetDiff(key K) (cloudclient.ConnectionsCount, bool) {
+	currentCounter, hasCurrentValue := c.currentCounters[key]
+	prevCounter, hasPrevValue := c.previousCounters[key]
+
+	if hasCurrentValue && !hasPrevValue {
+		connectionsCount, isValid := currentCounter.GetConnectionCount()
+		if isValid {
+			return cloudclient.ConnectionsCount{
+				Current: lo.ToPtr(connectionsCount),
+				Added:   lo.ToPtr(connectionsCount),
+				Removed: lo.ToPtr(0),
+			}, true
+		}
+		return cloudclient.ConnectionsCount{}, false
+	}
+
+	if !hasCurrentValue && hasPrevValue {
+		connectionsCount, isValid := prevCounter.GetConnectionCount()
+		if isValid {
+			return cloudclient.ConnectionsCount{
+				Current: lo.ToPtr(0),
+				Added:   lo.ToPtr(0),
+				Removed: lo.ToPtr(connectionsCount),
+			}, true
+		}
+		return cloudclient.ConnectionsCount{}, false
+	}
+
+	if hasCurrentValue && hasPrevValue {
+		connectionDiff, valid := currentCounter.GetConnectionCountDiff(prevCounter)
+		if valid {
+			return connectionDiff, true
+		}
+	}
+
+	return cloudclient.ConnectionsCount{}, false
+}

src/mapper/pkg/concurrentconnectioncounter/connection_count_differ.go

@@ -0,0 +1,140 @@
+package concurrentconnectioncounter
+
+import (
+	"github.com/stretchr/testify/suite"
+	"testing"
+)
+
+type ConnectionCountDifferSuite struct {
+	suite.Suite
+}
+
+type CountableIntentTCPDummy struct{}
+type CountableIntentDNSDummy struct{}
+
+func NewCountableIntentDummy() *CountableIntentTCPDummy {
+	return &CountableIntentTCPDummy{}
+}
+
+func (c *CountableIntentTCPDummy) ShouldCountUsingSrcPortMethod() bool {
+	return true
+}
+
+func (c *CountableIntentTCPDummy) ShouldCountUsingDNSMethod() bool {
+	return false
+}
+
+func NewCountableIntentDNSDummy() *CountableIntentDNSDummy {
+	return &CountableIntentDNSDummy{}
+}
+
+func (c *CountableIntentDNSDummy) ShouldCountUsingSrcPortMethod() bool {
+	return false
+}
+
+func (c *CountableIntentDNSDummy) ShouldCountUsingDNSMethod() bool {
+	return true
+}
+
+func (s *ConnectionCountDifferSuite) TestTCPDiff_TestNoPrevValue() {
+	differ := NewConnectionCountDiffer[string, *CountableIntentTCPDummy]()
+
+	// Add a connections
+	differ.Increment("key1", CounterInput[*CountableIntentTCPDummy]{
+		Intent:      NewCountableIntentDummy(),
+		SourcePorts: []int64{1, 2},
+	})
+	differ.Increment("key1", CounterInput[*CountableIntentTCPDummy]{
+		Intent:      NewCountableIntentDummy(),
+		SourcePorts: []int64{2, 3},
+	})
+
+	// Get the diff
+	diff, ok := differ.GetDiff("key1")
+
+	s.Require().True(ok)
+	s.Require().Equal(3, *diff.Current)
+	s.Require().Equal(3, *diff.Added)
+	s.Require().Equal(0, *diff.Removed)
+}
+
+func (s *ConnectionCountDifferSuite) TestTCPDiff_TestPrevConnectionsAreTheSameAsCurrent() {
+	differ := NewConnectionCountDiffer[string, *CountableIntentTCPDummy]()
+
+	// Add a connections
+	differ.Increment("key1", CounterInput[*CountableIntentTCPDummy]{
+		Intent:      NewCountableIntentDummy(),
+		SourcePorts: []int64{1, 2, 3},
+	})
+
+	differ.Reset()
+
+	// Add same connections
+	differ.Increment("key1", CounterInput[*CountableIntentTCPDummy]{
+		Intent:      NewCountableIntentDummy(),
+		SourcePorts: []int64{1, 2, 3},
+	})
+
+	// Get the diff
+	diff, ok := differ.GetDiff("key1")
+
+	s.Require().True(ok)
+	s.Require().Equal(3, *diff.Current)
+	s.Require().Equal(0, *diff.Added)
+	s.Require().Equal(0, *diff.Removed)
+}
+
+func (s *ConnectionCountDifferSuite) TestDNSDiff_TestNoPrevValue() {
+	differ := NewConnectionCountDiffer[string, *CountableIntentDNSDummy]()
+
+	// Add a connections
+	differ.Increment("key1", CounterInput[*CountableIntentDNSDummy]{
+		Intent:      NewCountableIntentDNSDummy(),
+		SourcePorts: []int64{1, 2},
+	})
+	differ.Increment("key1", CounterInput[*CountableIntentDNSDummy]{
+		Intent:      NewCountableIntentDNSDummy(),
+		SourcePorts: []int64{2, 3},
+	})
+
+	// Get the diff
+	diff, ok := differ.GetDiff("key1")
+
+	s.Require().True(ok)
+	s.Require().Equal(2, *diff.Current)
+	s.Require().Equal(2, *diff.Added)
+	s.Require().Equal(0, *diff.Removed)
+}
+
+func (s *ConnectionCountDifferSuite) TestDNSDiff_TestWithPrevValue() {
+	differ := NewConnectionCountDiffer[string, *CountableIntentDNSDummy]()
+
+	// Add a connections
+	differ.Increment("key1", CounterInput[*CountableIntentDNSDummy]{
+		Intent:      NewCountableIntentDNSDummy(),
+		SourcePorts: []int64{1, 2},
+	})
+	differ.Increment("key1", CounterInput[*CountableIntentDNSDummy]{
+		Intent:      NewCountableIntentDNSDummy(),
+		SourcePorts: []int64{2, 3},
+	})
+
+	differ.Reset()
+
+	differ.Increment("key1", CounterInput[*CountableIntentDNSDummy]{
+		Intent:      NewCountableIntentDNSDummy(),
+		SourcePorts: make([]int64, 0),
+	})
+
+	// Get the diff
+	diff, ok := differ.GetDiff("key1")
+
+	s.Require().True(ok)
+	s.Require().Equal(1, *diff.Current)
+	s.Require().Equal(1, *diff.Added)
+	s.Require().Equal(2, *diff.Removed)
+}
+
+func TestConnectionCountDifferSuite(t *testing.T) {
+	suite.Run(t, new(ConnectionCountDifferSuite))
+}