Adding Compression and Deduplication logic

pkg/sloop/common/logging.go

@@ -1,4 +1,3 @@
-
 package common
 
 import "github.com/golang/glog"

pkg/sloop/ingress/kubewatcher_test.go

@@ -156,22 +156,22 @@ func Test_bigPictureWithExclusionRules(t *testing.T) {
 		t.Fatalf("Error creating service: %v\n", err)
 	}
 
+	expectedEvents := 3
 	eventCount := 0
-loop:
-	for {
+	deadline := time.After(10 * time.Second)
+	for eventCount < expectedEvents {
 		select {
-		case <-time.After(1 * time.Second):
-			break loop
+		case <-deadline:
+			t.Fatalf("Timed out waiting for events: got %d, expected %d", eventCount, expectedEvents)
 		case result, ok := <-outChan:
-			if ok {
-				eventCount++
-				assert.NotContains(t, result.Payload, `"name":"s2"`)
-			} else {
-				t.Fatalf("Channel closed unexpectedly: %v\n", ok)
+			if !ok {
+				t.Fatalf("Channel closed unexpectedly")
 			}
+			eventCount++
+			assert.NotContains(t, result.Payload, `"name":"s2"`)
 		}
 	}
-	assert.Equal(t, 3, eventCount) // assert no event for service named s2
+	assert.Equal(t, expectedEvents, eventCount)
 
 	kw.Stop()
 }

pkg/sloop/kubeextractor/payloadhash.go

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2019, salesforce.com, inc.
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ * For full license text, see LICENSE.txt file in the repo root or https://opensource.org/licenses/BSD-3-Clause
+ */
+
+package kubeextractor
+
+import (
+	"hash/fnv"
+
+	"github.com/Jeffail/gabs/v2"
+	"github.com/pkg/errors"
+)
+
+// VolatileFields are JSON paths stripped before hashing for dedup.
+// metadata volatile fields change on every API server touch but don't represent real changes.
+// The entire "status" block is stripped because status changes constantly (replicas, metrics,
+// container states) but spec/labels/annotations changes are what matter for debugging.
+// Full payloads (including status) are still written on every actual write and every 30m snapshot.
+var VolatileFields = [][]string{
+	{"metadata", "resourceVersion"},
+	{"metadata", "generation"},
+	{"metadata", "uid"},
+	{"metadata", "selfLink"},
+	{"metadata", "managedFields"},
+	{"status"},
+}
+
+// ComputePayloadHash computes an FNV-64a hash of the payload after stripping volatile fields.
+// Returns uint64 hash suitable for fast in-memory comparison.
+func ComputePayloadHash(payload string) (uint64, error) {
+	cleanPayload, err := stripVolatileFields(payload)
+	if err != nil {
+		return 0, err
+	}
+
+	h := fnv.New64a()
+	_, err = h.Write([]byte(cleanPayload))
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to hash payload")
+	}
+
+	return h.Sum64(), nil
+}
+
+// stripVolatileFields removes known volatile/noisy fields from a Kubernetes resource JSON
+// This ensures that the hash is stable across frequent updates that don't represent real changes
+func stripVolatileFields(payload string) (string, error) {
+	jsonParsed, err := gabs.ParseJSON([]byte(payload))
+	if err != nil {
+		return "", errors.Wrap(err, "failed to parse JSON payload")
+	}
+
+	// Strip each volatile field path, ignoring errors if the path doesn't exist
+	for _, fieldPath := range VolatileFields {
+		_ = jsonParsed.Delete(fieldPath...)
+	}
+
+	return jsonParsed.String(), nil
+}

pkg/sloop/kubeextractor/payloadhash_test.go

@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2019, salesforce.com, inc.
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ * For full license text, see LICENSE.txt file in the repo root or https://opensource.org/licenses/BSD-3-Clause
+ */
+
+package kubeextractor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestComputePayloadHash_StableForSamePayload(t *testing.T) {
+	payload := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default",
+    "resourceVersion": "12345"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  }
+}`
+
+	hash1, err := ComputePayloadHash(payload)
+	assert.NoError(t, err)
+
+	hash2, err := ComputePayloadHash(payload)
+	assert.NoError(t, err)
+
+	assert.Equal(t, hash1, hash2, "Hash should be stable for identical payloads")
+}
+
+func TestComputePayloadHash_IgnoresResourceVersion(t *testing.T) {
+	payload1 := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default",
+    "resourceVersion": "12345"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  }
+}`
+
+	payload2 := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default",
+    "resourceVersion": "99999"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  }
+}`
+
+	hash1, err := ComputePayloadHash(payload1)
+	assert.NoError(t, err)
+
+	hash2, err := ComputePayloadHash(payload2)
+	assert.NoError(t, err)
+
+	assert.Equal(t, hash1, hash2, "Hash should ignore resourceVersion changes")
+}
+
+func TestComputePayloadHash_IgnoresStatusConditions(t *testing.T) {
+	payload1 := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  },
+  "status": {
+    "phase": "Running",
+    "conditions": [
+      {
+        "type": "Ready",
+        "status": "True",
+        "lastTransitionTime": "2026-04-01T10:00:00Z"
+      }
+    ]
+  }
+}`
+
+	payload2 := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  },
+  "status": {
+    "phase": "Running",
+    "conditions": [
+      {
+        "type": "Ready",
+        "status": "True",
+        "lastTransitionTime": "2026-04-02T15:30:00Z"
+      }
+    ]
+  }
+}`
+
+	hash1, err := ComputePayloadHash(payload1)
+	assert.NoError(t, err)
+
+	hash2, err := ComputePayloadHash(payload2)
+	assert.NoError(t, err)
+
+	assert.Equal(t, hash1, hash2, "Hash should ignore status conditions and timestamps")
+}
+
+func TestComputePayloadHash_DetectsMeaningfulChanges(t *testing.T) {
+	payloadBefore := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v1"
+      }
+    ]
+  }
+}`
+
+	payloadAfter := `{
+  "metadata": {
+    "name": "test-pod",
+    "namespace": "default"
+  },
+  "spec": {
+    "containers": [
+      {
+        "name": "app",
+        "image": "app:v2"
+      }
+    ]
+  }
+}`
+
+	hashBefore, err := ComputePayloadHash(payloadBefore)
+	assert.NoError(t, err)
+
+	hashAfter, err := ComputePayloadHash(payloadAfter)
+	assert.NoError(t, err)
+
+	assert.NotEqual(t, hashBefore, hashAfter, "Hash should differ for meaningful spec changes")
+}
+
+func TestComputePayloadHash_InvalidJSON(t *testing.T) {
+	invalidPayload := `{invalid json`
+
+	_, err := ComputePayloadHash(invalidPayload)
+	assert.Error(t, err, "Should error on invalid JSON")
+}
+
+func TestStripVolatileFields_RemovesResourceVersion(t *testing.T) {
+	payload := `{
+  "metadata": {
+    "name": "test",
+    "resourceVersion": "12345"
+  },
+  "spec": {
+    "image": "img:v1"
+  }
+}`
+
+	cleaned, err := stripVolatileFields(payload)
+	assert.NoError(t, err)
+	assert.NotContains(t, cleaned, "12345", "resourceVersion should be removed")
+	assert.Contains(t, cleaned, "test", "name should remain")
+	assert.Contains(t, cleaned, "img:v1", "spec.image should remain")
+}
+
+func TestStripVolatileFields_PreservesSpec(t *testing.T) {
+	payload := `{
+  "metadata": {
+    "name": "test-pod"
+  },
+  "spec": {
+    "replicas": 3,
+    "template": {
+      "spec": {
+        "containers": [
+          {
+            "name": "app",
+            "image": "app:latest"
+          }
+        ]
+      }
+    }
+  }
+}`
+
+	cleaned, err := stripVolatileFields(payload)
+	assert.NoError(t, err)
+	assert.Contains(t, cleaned, "replicas", "spec.replicas should be preserved")
+	assert.Contains(t, cleaned, "app:latest", "container image should be preserved")
+}