feat(inspection): add NodeNameDiscoveryTask for k8snode and k8scontainer log parsers

pkg/task/inspection/googlecloudlogcomputeapiaudit/impl/query_task.go

@@ -34,7 +34,7 @@ import (
 func GenerateComputeAPIQuery(taskMode inspectioncore_contract.InspectionTaskModeType, nodeNames []string) []string {
 	if taskMode == inspectioncore_contract.TaskModeDryRun {
 		return []string{
-			generateComputeAPIQueryWithInstanceNameFilter("-- instance name filters to be determined after audit log query"),
+			generateComputeAPIQueryWithInstanceNameFilter("-- instance name filters to be determined after node name discovery"),
 		}
 	} else {
 		result := []string{}

pkg/task/inspection/googlecloudlogcomputeapiaudit/impl/query_task_test.go

@@ -35,7 +35,7 @@ func TestGenerateComputeAPIQuery(t *testing.T) {
 			NodeNames: []string{}, // No nodes specified for dry run
 			Expected: []string{`resource.type="gce_instance"
 -protoPayload.methodName:("list" OR "get" OR "watch")
--- instance name filters to be determined after audit log query
+-- instance name filters to be determined after node name discovery
 `},
 		},
 		{

pkg/task/inspection/googlecloudlogk8scontainer/contract/taskid.go

@@ -53,3 +53,6 @@ var PodPhaseTimelineMapperTaskID = taskid.NewDefaultImplementationID[inspectiont
 
 // TailTaskID is a nop task just to require all container log mappers.
 var TailTaskID = taskid.NewDefaultImplementationID[struct{}](TaskIDPrefix + "tail")
+
+// NodeNameDiscoveryTaskID is the discovery task ID for extracting node names from Container log labels.
+var NodeNameDiscoveryTaskID = taskid.NewDefaultImplementationID[[]string](TaskIDPrefix + "node-name-discovery")

pkg/task/inspection/googlecloudlogk8scontainer/impl/nodename_discovery_task.go

@@ -0,0 +1,55 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package googlecloudlogk8scontainer_impl
+
+import (
+	"context"
+
+	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
+	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
+	"github.com/GoogleCloudPlatform/khi/pkg/model/log"
+	commonlogk8saudit_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/commonlogk8saudit/contract"
+	googlecloudlogk8scontainer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudlogk8scontainer/contract"
+	inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
+)
+
+// NodeNameDiscoveryTask extracts node names from Kubernetes Container log labels and registers them to NodeNameInventoryTask.
+var NodeNameDiscoveryTask = commonlogk8saudit_contract.NodeNameInventoryBuilder.DiscoveryTask(
+	googlecloudlogk8scontainer_contract.NodeNameDiscoveryTaskID,
+	[]taskid.UntypedTaskReference{
+		googlecloudlogk8scontainer_contract.FieldSetReaderTaskID.Ref(),
+	},
+	func(ctx context.Context, taskMode inspectioncore_contract.InspectionTaskModeType, progress *inspectionmetadata.TaskProgressMetadata) ([]string, error) {
+		if taskMode == inspectioncore_contract.TaskModeDryRun {
+			return nil, nil
+		}
+
+		foundNodeNames := map[string]struct{}{}
+		logs := coretask.GetTaskResult(ctx, googlecloudlogk8scontainer_contract.FieldSetReaderTaskID.Ref())
+		for _, l := range logs {
+			fs, err := log.GetFieldSet(l, &googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{})
+			if err == nil && fs.NodeName != "" {
+				foundNodeNames[fs.NodeName] = struct{}{}
+			}
+		}
+
+		var result []string
+		for k := range foundNodeNames {
+			result = append(result, k)
+		}
+		return result, nil
+	},
+)

pkg/task/inspection/googlecloudlogk8scontainer/impl/nodename_discovery_task_test.go

@@ -0,0 +1,98 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package googlecloudlogk8scontainer_impl
+
+import (
+	"sort"
+	"testing"
+
+	inspectiontest "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/test"
+	tasktest "github.com/GoogleCloudPlatform/khi/pkg/core/task/test"
+	"github.com/GoogleCloudPlatform/khi/pkg/model/log"
+	googlecloudlogk8scontainer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudlogk8scontainer/contract"
+	inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestNodeNameDiscoveryTask(t *testing.T) {
+	tests := []struct {
+		name     string
+		logs     []*log.Log
+		taskMode inspectioncore_contract.InspectionTaskModeType
+		want     []string
+	}{
+		{
+			name: "valid container logs with node name labels",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-2",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeRun,
+			want: []string{
+				"gke-test-cluster-default-pool-node-1",
+				"gke-test-cluster-default-pool-node-2",
+			},
+		},
+		{
+			name: "empty node name label is ignored",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeRun,
+			want: []string{
+				"gke-test-cluster-default-pool-node-1",
+			},
+		},
+		{
+			name: "dry run returns nil",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8scontainer_contract.GCPContainerLogNodeNameLabelFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeDryRun,
+			want:     nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := inspectiontest.WithDefaultTestInspectionTaskContext(t.Context())
+			result, _, err := inspectiontest.RunInspectionTask(ctx, NodeNameDiscoveryTask, tc.taskMode, map[string]any{},
+				tasktest.NewTaskDependencyValuePair(googlecloudlogk8scontainer_contract.FieldSetReaderTaskID.Ref(), tc.logs),
+			)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			sort.Strings(result)
+			sort.Strings(tc.want)
+			if diff := cmp.Diff(tc.want, result); diff != "" {
+				t.Errorf("result mismatch (-want +got):\n%s", diff)
+			}
+		})
+	}
+}

pkg/task/inspection/googlecloudlogk8scontainer/impl/parser_task.go

@@ -335,6 +335,7 @@ var TailTask = inspectiontaskbase.NewInspectionTask(
 	[]taskid.UntypedTaskReference{
 		googlecloudlogk8scontainer_contract.LogToTimelineMapperTaskID.Ref(),
 		googlecloudlogk8scontainer_contract.PodPhaseTimelineMapperTaskID.Ref(),
+		googlecloudlogk8scontainer_contract.NodeNameDiscoveryTaskID.Ref(),
 	},
 	func(ctx context.Context, taskMode inspectioncore_contract.InspectionTaskModeType) (struct{}, error) {
 		return struct{}{}, nil

pkg/task/inspection/googlecloudlogk8scontainer/impl/registration.go

@@ -50,5 +50,6 @@ func Register(registry coreinspection.InspectionTaskRegistry) error {
 		LogToTimelineMapperTask,
 		PodPhaseTimelineMapperTask,
 		TailTask,
+		NodeNameDiscoveryTask,
 	)
 }

pkg/task/inspection/googlecloudlogk8snode/contract/taskid.go

@@ -75,3 +75,6 @@ var OtherLogLogToTimelineMapperTaskID = taskid.NewDefaultImplementationID[inspec
 var TailTaskID = taskid.NewDefaultImplementationID[struct{}](TaskIDPrefix + "tail")
 
 var ContainerIDDiscoveryTaskID = taskid.NewDefaultImplementationID[commonlogk8saudit_contract.ContainerIDToContainerIdentity](TaskIDPrefix + "container-id-discovery")
+
+// NodeNameDiscoveryTaskID is the task ID for extracting node names from Kubernetes node logs.
+var NodeNameDiscoveryTaskID = taskid.NewDefaultImplementationID[[]string](TaskIDPrefix + "node-name-discovery")

pkg/task/inspection/googlecloudlogk8snode/impl/common_task.go

@@ -191,6 +191,7 @@ var TailTask = inspectiontaskbase.NewInspectionTask(googlecloudlogk8snode_contra
 		googlecloudlogk8snode_contract.OtherLogLogToTimelineMapperTaskID.Ref(),
 
 		googlecloudlogk8snode_contract.ContainerIDDiscoveryTaskID.Ref(),
+		googlecloudlogk8snode_contract.NodeNameDiscoveryTaskID.Ref(),
 	},
 	func(ctx context.Context, taskMode inspectioncore_contract.InspectionTaskModeType) (struct{}, error) {
 		return struct{}{}, nil

pkg/task/inspection/googlecloudlogk8snode/impl/nodename_discovery_task.go

@@ -0,0 +1,55 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package googlecloudlogk8snode_impl
+
+import (
+	"context"
+
+	inspectionmetadata "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/metadata"
+	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
+	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
+	"github.com/GoogleCloudPlatform/khi/pkg/model/log"
+	commonlogk8saudit_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/commonlogk8saudit/contract"
+	googlecloudlogk8snode_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudlogk8snode/contract"
+	inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
+)
+
+// NodeNameDiscoveryTask extracts node names from Kubernetes Node component logs and registers them to NodeNameInventoryTask.
+var NodeNameDiscoveryTask = commonlogk8saudit_contract.NodeNameInventoryBuilder.DiscoveryTask(
+	googlecloudlogk8snode_contract.NodeNameDiscoveryTaskID,
+	[]taskid.UntypedTaskReference{
+		googlecloudlogk8snode_contract.CommonFieldsetReaderTaskID.Ref(),
+	},
+	func(ctx context.Context, taskMode inspectioncore_contract.InspectionTaskModeType, progress *inspectionmetadata.TaskProgressMetadata) ([]string, error) {
+		if taskMode == inspectioncore_contract.TaskModeDryRun {
+			return nil, nil
+		}
+
+		foundNodeNames := map[string]struct{}{}
+		logs := coretask.GetTaskResult(ctx, googlecloudlogk8snode_contract.CommonFieldsetReaderTaskID.Ref())
+		for _, l := range logs {
+			fs, err := log.GetFieldSet(l, &googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{})
+			if err == nil && fs.NodeName != "" {
+				foundNodeNames[fs.NodeName] = struct{}{}
+			}
+		}
+
+		var result []string
+		for k := range foundNodeNames {
+			result = append(result, k)
+		}
+		return result, nil
+	},
+)

pkg/task/inspection/googlecloudlogk8snode/impl/nodename_discovery_task_test.go

@@ -0,0 +1,98 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package googlecloudlogk8snode_impl
+
+import (
+	"sort"
+	"testing"
+
+	inspectiontest "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/test"
+	tasktest "github.com/GoogleCloudPlatform/khi/pkg/core/task/test"
+	"github.com/GoogleCloudPlatform/khi/pkg/model/log"
+	googlecloudlogk8snode_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudlogk8snode/contract"
+	inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestNodeNameDiscoveryTask(t *testing.T) {
+	tests := []struct {
+		name     string
+		logs     []*log.Log
+		taskMode inspectioncore_contract.InspectionTaskModeType
+		want     []string
+	}{
+		{
+			name: "valid node logs with node names",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-2",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeRun,
+			want: []string{
+				"gke-test-cluster-default-pool-node-1",
+				"gke-test-cluster-default-pool-node-2",
+			},
+		},
+		{
+			name: "empty node name field is ignored",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "",
+				}),
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeRun,
+			want: []string{
+				"gke-test-cluster-default-pool-node-1",
+			},
+		},
+		{
+			name: "dry run returns nil",
+			logs: []*log.Log{
+				log.NewLogWithFieldSetsForTest(&googlecloudlogk8snode_contract.K8sNodeLogCommonFieldSet{
+					NodeName: "gke-test-cluster-default-pool-node-1",
+				}),
+			},
+			taskMode: inspectioncore_contract.TaskModeDryRun,
+			want:     nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := inspectiontest.WithDefaultTestInspectionTaskContext(t.Context())
+			result, _, err := inspectiontest.RunInspectionTask(ctx, NodeNameDiscoveryTask, tc.taskMode, map[string]any{},
+				tasktest.NewTaskDependencyValuePair(googlecloudlogk8snode_contract.CommonFieldsetReaderTaskID.Ref(), tc.logs),
+			)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			sort.Strings(result)
+			sort.Strings(tc.want)
+			if diff := cmp.Diff(tc.want, result); diff != "" {
+				t.Errorf("result mismatch (-want +got):\n%s", diff)
+			}
+		})
+	}
+}

pkg/task/inspection/googlecloudlogk8snode/impl/registration.go

@@ -138,5 +138,6 @@ func Register(registry coreinspection.InspectionTaskRegistry) error {
 		OtherLogLogToTimelineMapperTask,
 		TailTask,
 		ContainerIDDiscoveryTask,
+		NodeNameDiscoveryTask,
 	)
 }

pkg/task/inspection/googlecloudlogserialport/impl/query_task.go

@@ -35,7 +35,7 @@ const MaxNodesPerQuery = 30
 func GenerateSerialPortQuery(taskMode inspectioncore_contract.InspectionTaskModeType, foundNodeNames []string, nodeNameSubstrings []string) []string {
 	if taskMode == inspectioncore_contract.TaskModeDryRun {
 		return []string{
-			generateSerialPortQueryWithInstanceNameFilter("-- instance name filters to be determined after audit log query", generateNodeNameSubstringLogFilter(nodeNameSubstrings)),
+			generateSerialPortQueryWithInstanceNameFilter("-- instance name filters to be determined after node name discovery", generateNodeNameSubstringLogFilter(nodeNameSubstrings)),
 		}
 	} else {
 		result := []string{}

pkg/task/inspection/googlecloudlogserialport/impl/query_task_test.go

@@ -42,7 +42,7 @@ LOG_ID("serialconsole.googleapis.com%2Fserial_port_2_output") OR
 LOG_ID("serialconsole.googleapis.com%2Fserial_port_3_output") OR
 LOG_ID("serialconsole.googleapis.com%2Fserial_port_debug_output")
 
--- instance name filters to be determined after audit log query
+-- instance name filters to be determined after node name discovery
 
 -- No node name substring filters are specified.`,
 		},