Stop ingesting initial state

Author: TylerHelmuth

receiver/k8sobjectsreceiver/receiver.go

@@ -147,25 +147,11 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 	kr.stopperChanList = append(kr.stopperChanList, stopperChan)
 	kr.mu.Unlock()
 
-	resourceVersion := config.ResourceVersion
-	var err error
-	if resourceVersion == "" {
-		// Proper use of the Kubernetes API Watch capability when no resourceVersion is supplied is to do a list first
-		// to get the initial state and a useable resourceVersion.
-		// See https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes for details.
-		resourceVersion, err = kr.doInitialList(ctx, config, resource)
-		if err != nil {
-			kr.setting.Logger.Error("could not perform initial list for watch", zap.String("resource", config.gvr.String()), zap.Error(err))
-			return
-		}
-		// If we still don't have a resourceVersion we can try 1 as a last ditch effort.
-		// This also helps our unit tests since the fake client can't handle returning resource versions
-		// as part of a list of objects.
-		if resourceVersion == "" {
-			resourceVersion = defaultResourceVersion
-		}
+	resourceVersion, err := getResourceVersion(ctx, config, resource)
+	if err != nil {
+		kr.setting.Logger.Error("could not retrieve an initial resourceVersion", zap.String("resource", config.gvr.String()), zap.Error(err))
+		return
 	}
-
 	watchFunc := func(options metav1.ListOptions) (apiWatch.Interface, error) {
 		options.FieldSelector = config.FieldSelector
 		options.LabelSelector = config.LabelSelector
@@ -202,28 +188,33 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 
 }
 
-func (kr *k8sobjectsreceiver) doInitialList(ctx context.Context, config *K8sObjectsConfig, resource dynamic.ResourceInterface) (string, error) {
-	objects, err := resource.List(ctx, metav1.ListOptions{
-		FieldSelector: config.FieldSelector,
-		LabelSelector: config.LabelSelector,
-	})
-	if err != nil {
-		return "", err
-	}
+func getResourceVersion(ctx context.Context, config *K8sObjectsConfig, resource dynamic.ResourceInterface) (string, error) {
+	resourceVersion := config.ResourceVersion
+	if resourceVersion == "" || resourceVersion == "0" {
+		// Proper use of the Kubernetes API Watch capability when no resourceVersion is supplied is to do a list first
+		// to get the initial state and a useable resourceVersion.
+		// See https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes for details.
+		objects, err := resource.List(ctx, metav1.ListOptions{
+			FieldSelector: config.FieldSelector,
+			LabelSelector: config.LabelSelector,
+		})
+		if err != nil {
+			return "", fmt.Errorf("could not perform initial list for watch on %v, %w", config.gvr.String(), err)
+		}
+		if objects == nil {
+			return "", fmt.Errorf("nil objects returned, this is an error in the k8sobjectsreceiver")
+		}
 
-	if objects == nil {
-		return "", fmt.Errorf("nil objects returned, this is an error in the k8sobjectsreceiver")
-	}
+		resourceVersion = objects.GetResourceVersion()
 
-	if len(objects.Items) > 0 {
-		logs := pullObjectsToLogData(objects, time.Now(), config)
-		obsCtx := kr.obsrecv.StartLogsOp(ctx)
-		err = kr.consumer.ConsumeLogs(obsCtx, logs)
-		kr.obsrecv.EndLogsOp(obsCtx, metadata.Type, logs.LogRecordCount(), err)
+		// If we still don't have a resourceVersion we can try 1 as a last ditch effort.
+		// This also helps our unit tests since the fake client can't handle returning resource versions
+		// as part of a list of objects.
+		if resourceVersion == "" || resourceVersion == "0" {
+			resourceVersion = defaultResourceVersion
+		}
 	}
-
-	return objects.GetResourceVersion(), nil
-
+	return resourceVersion, nil
 }
 
 // Start ticking immediately.

receiver/k8sobjectsreceiver/receiver_test.go

@@ -88,12 +88,6 @@ func TestWatchObject(t *testing.T) {
 		generatePod("pod1", "default", map[string]interface{}{
 			"environment": "production",
 		}, "1"),
-		generatePod("pod2", "default", map[string]interface{}{
-			"environment": "test",
-		}, "2"),
-		generatePod("pod3", "default_ignore", map[string]interface{}{
-			"environment": "production",
-		}, "3"),
 	)
 
 	rCfg := createDefaultConfig().(*Config)
@@ -124,17 +118,23 @@ func TestWatchObject(t *testing.T) {
 	require.NoError(t, r.Start(ctx, componenttest.NewNopHost()))
 
 	time.Sleep(time.Millisecond * 100)
-	assert.Len(t, consumer.Logs(), 1)
-	assert.Equal(t, 2, consumer.Count())
+	assert.Len(t, consumer.Logs(), 0)
+	assert.Equal(t, 0, consumer.Count())
 
 	mockClient.createPods(
+		generatePod("pod2", "default", map[string]interface{}{
+			"environment": "test",
+		}, "2"),
+		generatePod("pod3", "default_ignore", map[string]interface{}{
+			"environment": "production",
+		}, "3"),
 		generatePod("pod4", "default", map[string]interface{}{
 			"environment": "production",
 		}, "4"),
 	)
 	time.Sleep(time.Millisecond * 100)
 	assert.Len(t, consumer.Logs(), 2)
-	assert.Equal(t, 3, consumer.Count())
+	assert.Equal(t, 2, consumer.Count())
 
 	assert.NoError(t, r.Shutdown(ctx))
 }