k8snode_impl

var ClusterIdentityAliasTask = coretask.NewAliasTask(
	k8snode.ClusterIdentityTaskID,
	k8scommon.ClusterIdentityTaskID.Ref(),
)

var ContainerIDDiscoveryTask = inspectiontaskbase.NewInspectionTask(k8snode.ContainerIDDiscoveryTaskID,
	[]coretask.Dependency{
		k8snode.ContainerdLogFilterTaskID.Ref(),
	},
	func(ctx context.Context, taskMode inspectioncore.InspectionTaskModeType) (k8saudit.ContainerIDToContainerIdentity, error) {
		if taskMode == inspectioncore.TaskModeDryRun {
			return nil, nil
		}

		logs := coretask.GetTaskResult(ctx, k8snode.ContainerdLogFilterTaskID.Ref())

		tracker := progress.NewTracker(ctx, len(logs), progress.WithUnit("logs"))
		defer tracker.Done()

		result := k8saudit.ContainerIDToContainerIdentity{}
		logChan := make(chan *log.Log)
		errGrp, childRoutineCtx := errgroup.WithContext(ctx)
		containerIdentitiesChan := make(chan *k8saudit.ContainerIdentity, runtime.GOMAXPROCS(0))
		for i := 0; i < runtime.GOMAXPROCS(0); i++ {
			errGrp.Go(func() error {
				for {
					select {
					case <-childRoutineCtx.Done():
						return childRoutineCtx.Err()
					case l, ok := <-logChan:
						if !ok {
							return nil
						}
						processContainerIDDiscoveryForLog(ctx, l, containerIdentitiesChan)
						tracker.Inc()
					}
				}
			})
		}
		consumerGrp, childConsumerRoutineCtx := errgroup.WithContext(ctx)
		consumerGrp.Go(func() error {
			for {
				select {
				case <-childConsumerRoutineCtx.Done():
					return childConsumerRoutineCtx.Err()
				case c, ok := <-containerIdentitiesChan:
					if !ok {
						return nil
					}
					result[c.ContainerID] = c
				}
			}
		})

		for _, l := range logs {
			logChan <- l
		}
		close(logChan)
		err := errGrp.Wait()
		close(containerIdentitiesChan)
		consumerErr := consumerGrp.Wait()
		if err != nil {
			return nil, err
		}
		if consumerErr != nil {
			return nil, consumerErr
		}

		return result, nil
	},
	coretask.ProvidesTag(k8saudit.TagContainerIDDiscovery),
)

var ContainerdLogFilterTask = newParserTypeFilterTask(k8snode.ContainerdLogFilterTaskID, k8snode.ListLogEntriesTaskID.Ref(), k8snode.Containerd)

var ContainerdLogGroupTask = newNodeAndComponentNameGrouperTask(k8snode.ContainerdLogGroupTaskID, k8snode.ContainerdLogFilterTaskID.Ref())

var ContainerdNodeLogLogToTimelineMapperTask = inspectiontaskbase.NewLogToTimelineMapperTask(
	k8snode.ContainerdLogLogToTimelineMapperTaskID,
	&containerdNodeLogLogToTimelineMapperSetting{},
)

var KubeletLogFilterTask = newParserTypeFilterTask(k8snode.KubeletLogFilterTaskID, k8snode.ListLogEntriesTaskID.Ref(), k8snode.Kubelet)

var KubeletLogGroupTask = newNodeAndComponentNameGrouperTask(k8snode.KubeletLogGroupTaskID, k8snode.KubeletLogFilterTaskID.Ref())

var KubeletLogLogToTimelineMapperTask = inspectiontaskbase.NewLogToTimelineMapperTask(
	k8snode.KubeletLogLogToTimelineMapperTaskID,
	&kubeletNodeLogLogToTimelineMapperSetting{},
)

var ListLogEntriesTask = gcpcommon.NewStructuredListLogEntriesTask(&k8snodeListLogEntriesTaskSetting{})

var LogIngesterTask = inspectiontaskbase.NewLogIngesterTask(
	k8snode.LogIngesterTaskID,
	&K8sNodeLogIngester{},
)

var NodeNameDiscoveryTask = inspectiontaskbase.NewInspectionTask(
	k8snode.NodeNameDiscoveryTaskID,
	[]coretask.Dependency{
		k8snode.ListLogEntriesTaskID.Ref(),
	},
	func(ctx context.Context, taskMode inspectioncore.InspectionTaskModeType) ([]string, error) {
		if taskMode == inspectioncore.TaskModeDryRun {
			return nil, nil
		}

		foundNodeNames := map[string]struct{}{}
		logs := coretask.GetTaskResult(ctx, k8snode.ListLogEntriesTaskID.Ref())
		for _, l := range logs {
			fs, err := k8snode.ExtractK8sNodeLogCommon(l.NodeReader, nil)
			if err == nil && fs.NodeName != "" {
				foundNodeNames[fs.NodeName] = struct{}{}
			}
		}

		var result []string
		for k := range foundNodeNames {
			result = append(result, k)
		}
		return result, nil
	},
	coretask.ProvidesTag(k8saudit.TagNodeNameDiscovery),
)

var OtherLogFilterTask = newParserTypeFilterTask(k8snode.OtherLogFilterTaskID, k8snode.ListLogEntriesTaskID.Ref(), k8snode.Other)

var OtherLogGroupTask = newNodeAndComponentNameGrouperTask(k8snode.OtherLogGroupTaskID, k8snode.OtherLogFilterTaskID.Ref())

var OtherLogLogToTimelineMapperTask = inspectiontaskbase.NewLogToTimelineMapperTask(k8snode.OtherLogLogToTimelineMapperTaskID, &otherNodeLogLogToTimelineMapperSetting{
	StartingMessagesByComponent: map[string]string{
		"dockerd":             "Starting up",
		"configure.sh":        "Start to install kubernetes files",
		"configure-helper.sh": "Start to configure instance for kubernetes",
	},
	TerminatingMessagesByComponent: map[string]string{
		"dockerd":             "Daemon shutdown complete",
		"configure.sh":        "Done for installing kubernetes files",
		"configure-helper.sh": "Done for the configuration for kubernetes",
	},
})

var PodSandboxIDDiscoveryTask = inspectiontaskbase.NewInspectionTask(k8snode.PodSandboxIDDiscoveryTaskID,
	[]coretask.Dependency{
		k8snode.ContainerdLogFilterTaskID.Ref(),
	},
	func(ctx context.Context, taskMode inspectioncore.InspectionTaskModeType) (patternfinder.PatternFinder[*k8snode.PodSandboxIDInfo], error) {
		if taskMode == inspectioncore.TaskModeDryRun {
			return nil, nil
		}
		logs := coretask.GetTaskResult(ctx, k8snode.ContainerdLogFilterTaskID.Ref())

		tracker := progress.NewTracker(ctx, len(logs), progress.WithUnit("logs"))
		defer tracker.Done()

		logChan := make(chan *log.Log)
		errGrp, childCtx := errgroup.WithContext(ctx)
		podSandboxIDFinder := patternfinder.NewRadixPatternFinder[*k8snode.PodSandboxIDInfo]()
		for i := 0; i < runtime.GOMAXPROCS(0); i++ {
			errGrp.Go(func() error {
				for {
					select {
					case <-childCtx.Done():
						return childCtx.Err()
					case l, ok := <-logChan:
						if !ok {
							return nil
						}
						processPodSandboxIDDiscoveryForLog(ctx, l, podSandboxIDFinder)
						tracker.Inc()
					}
				}
			})
		}

		for _, l := range logs {
			logChan <- l
		}
		close(logChan)
		errGrp.Wait()

		return podSandboxIDFinder, nil
	},
)

var TailTask = coretask.NewTailTask(
	k8snode.TailTaskID,
	[]coretask.Dependency{
		k8snode.ContainerdLogLogToTimelineMapperTaskID.Ref(),
		k8snode.KubeletLogLogToTimelineMapperTaskID.Ref(),
		k8snode.OtherLogLogToTimelineMapperTaskID.Ref(),

		k8snode.ContainerIDDiscoveryTaskID.Ref(),
		k8snode.NodeNameDiscoveryTaskID.Ref(),
	},
	inspectioncore.FeatureTaskLabel(
		"Kubernetes Node Logs",
		"Gather logs from Kubernetes node components (e.g., Docker, containerd, or Kubelet) to troubleshoot node-level issues. Note: The log volume can be very large if the cluster contains many nodes.",
		3000,
		false,
	),
)