ROX-20073: Implement Network Policy Store Reconciliation (stackrox#8153)

sensor/kubernetes/listener/resources/hash_reconciliation.go

@@ -87,6 +87,16 @@ func resourceToMessage(resType string, resID string) (*central.MsgFromSensor, er
 			},
 		}
 		return &central.MsgFromSensor{Msg: &msg}, nil
+	case deduper.TypeNetworkPolicy.String():
+		msg := central.MsgFromSensor_Event{
+			Event: &central.SensorEvent{
+				Id:     resID,
+				Action: central.ResourceAction_REMOVE_RESOURCE,
+				Resource: &central.SensorEvent_NetworkPolicy{
+					NetworkPolicy: &storage.NetworkPolicy{Id: resID}},
+			},
+		}
+		return &central.MsgFromSensor{Msg: &msg}, nil
 	case deduper.TypeNode.String():
 		msg := central.MsgFromSensor_Event{
 			Event: &central.SensorEvent{

sensor/kubernetes/listener/resources/hash_reconciliation_test.go

@@ -57,6 +57,11 @@ func (s *HashReconciliationSuite) TestResourceToMessage() {
 			expectedMsg:   &central.MsgFromSensor_Event{Event: &central.SensorEvent{Id: testResID, Action: central.ResourceAction_REMOVE_RESOURCE, Resource: &central.SensorEvent_Secret{Secret: &storage.Secret{Id: testResID}}}},
 			expectedError: nil,
 		},
+		"NetworkPolicy": {
+			resType:       deduper.TypeNetworkPolicy.String(),
+			expectedMsg:   &central.MsgFromSensor_Event{Event: &central.SensorEvent{Id: testResID, Action: central.ResourceAction_REMOVE_RESOURCE, Resource: &central.SensorEvent_NetworkPolicy{NetworkPolicy: &storage.NetworkPolicy{Id: testResID}}}},
+			expectedError: nil,
+		},
 		"Unknown should throw error": {
 			resType:       "Unknown",
 			expectedMsg:   nil,
@@ -99,6 +104,10 @@ func resourceTypeToFn(resType string) (func(*central.SensorEvent) string, error)
 		return func(event *central.SensorEvent) string {
 			return event.GetNode().GetId()
 		}, nil
+	case deduper.TypeNetworkPolicy.String():
+		return func(event *central.SensorEvent) string {
+			return event.GetNetworkPolicy().GetId()
+		}, nil
 	default:
 		return nil, errors.Errorf("not implemented for resource type %v", resType)
 	}
@@ -113,6 +122,8 @@ func initStore() *InMemoryStoreProvider {
 	s.podStore.addOrUpdatePod(&storage.Pod{Id: "4"})
 	s.nodeStore.addOrUpdateNode(makeNode("42"))
 	s.nodeStore.addOrUpdateNode(makeNode("43"))
+	s.networkPolicyStore.Upsert(&storage.NetworkPolicy{Id: "1"})
+	s.networkPolicyStore.Upsert(&storage.NetworkPolicy{Id: "2"})
 	s.serviceAccountStore.Add(&storage.ServiceAccount{
 		Id:               "5",
 		Name:             "Acc1",
@@ -265,6 +276,29 @@ func (s *HashReconciliationSuite) TestProcessHashes() {
 			},
 			deletedIDs: []string{"99", "98", "97"},
 		},
+		"No Network Policy": {
+			dstate: map[deduper.Key]uint64{
+				makeKey("1", deduper.TypeNetworkPolicy): 12345,
+				makeKey("2", deduper.TypeNetworkPolicy): 34567,
+			},
+			deletedIDs: []string{},
+		},
+		"Single Network Policy": {
+			dstate: map[deduper.Key]uint64{
+				makeKey("99", deduper.TypeNetworkPolicy): 34567,
+				makeKey("1", deduper.TypeNetworkPolicy):  12345,
+			},
+			deletedIDs: []string{"99"},
+		},
+		"Multiple Network Policies": {
+			dstate: map[deduper.Key]uint64{
+				makeKey("99", deduper.TypeNetworkPolicy): 34567,
+				makeKey("98", deduper.TypeNetworkPolicy): 34567,
+				makeKey("97", deduper.TypeNetworkPolicy): 34567,
+				makeKey("1", deduper.TypeNetworkPolicy):  12345,
+			},
+			deletedIDs: []string{"97", "98", "99"},
+		},
 	}
 
 	for n, c := range cases {

sensor/kubernetes/listener/resources/networkpolicy_store.go

@@ -4,6 +4,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/stackrox/rox/pkg/labels"
 	"github.com/stackrox/rox/pkg/sync"
+	"github.com/stackrox/rox/sensor/common/deduper"
 	"github.com/stackrox/rox/sensor/common/detector/metrics"
 	"github.com/stackrox/rox/sensor/common/store"
 
@@ -45,7 +46,7 @@ The Find operation returns all NetworkPolicies that would match a given set of l
 Example:
   policiesMatchingDeployment := store.Find("default"), map[string]string{"app": "nginx"})
 
-*) TODO: See ADR-XXX for alternative implementations that were considered
+See ADR-0002 "Design In-Memory Data Store for Network Policies in Sensor" for alternative implementations that were considered
 
 ## Complexities
 
@@ -86,10 +87,17 @@ type networkPolicyStoreImpl struct {
 // ReconcileDelete is called after Sensor reconnects with Central and receives its state hashes.
 // Reconciliacion ensures that Sensor and Central have the same state by checking whether a given resource
 // shall be deleted from Central.
-func (n *networkPolicyStoreImpl) ReconcileDelete(resType, resID string, resHash uint64) (string, error) {
-	_, _, _ = resType, resID, resHash
-	// TODO(ROX-20073): Implement me
-	return "", errors.New("Not implemented")
+func (n *networkPolicyStoreImpl) ReconcileDelete(resType, resID string, _ uint64) (string, error) {
+	if resType != deduper.TypeNetworkPolicy.String() {
+		return "", errors.Errorf("invalid resource type %v", resType)
+	}
+
+	p := n.Get(resID)
+	if p != nil {
+		return "", nil
+	}
+	// Resource on Central but not on Sensor, send for deletion
+	return resID, nil
 }
 
 func newNetworkPoliciesStore() *networkPolicyStoreImpl {

sensor/kubernetes/listener/resources/store_provider.go

@@ -83,6 +83,7 @@ func InitializeStore() *InMemoryStoreProvider {
 		deduper.TypeServiceAccount.String(): p.serviceAccountStore,
 		deduper.TypeSecret.String():         p.registryStore,
 		deduper.TypeNode.String():           p.nodeStore,
+		deduper.TypeNetworkPolicy.String():  p.networkPolicyStore,
 	}
 
 	return p