khi-parser

# KHI Log Parser Support Guidelines This guide outlines the patterns, package boundaries, implementation steps, and best practices for adding support for new log types or modifying existing log parsers in KHI. --- ## 1. Package Structure & Boundaries When implementing a new log parser or modifying an existing one, you MUST separate the **contract** (IDs, public types, and configurations) from the **implementation** (the actual task logic). This guarantees that task IDs are fully initialized before implementation and prevents circular import dependencies. The parser package must reside under `pkg/task/inspection/<provider>/<feature>/` (e.g., `pkg/task/inspection/googlecloud/k8snode/`) and adhere to the following structure: ```plaintext pkg/task/inspection/<provider>/<feature>/ ├── taskid.go // Defines all TaskIDs and TaskReferences (package <feature>). ├── extractor.go // (Optional) Defines field extraction functions and strongly-typed data structs. ├── timeline_type.go // (Optional) Defines timeline types and verb types. ├── timeline_path.go // (Optional) Helper functions to build hierarchical paths. ├── log_type.go // (Optional) Defines log-specific types or constants. └── impl/ // Package <feature>_impl ├── form.go // (Optional) Implements form-related parameter tasks. ├── query.go // (Optional) Implements log query/filter tasks. ├── ingester.go // (Optional) Implements the LogIngester task. ├── grouper.go // (Optional) Implements the LogGrouper task. ├── mapper.go // (Optional) Implements LogToTimelineMapper tasks (or mapper_<target>.go). └── module.go // Declares the package Module with its Scope and Tasks. ``` ### Key Package Boundaries > [!IMPORTANT] > > - **Contract / Root Package (`pkg/task/inspection/<provider>/<feature>`)**: Uses `package <feature>` (no `_contract` suffix). MUST NOT import the `impl` package. External packages can freely import the root package to depend on parser task IDs, Extractor functions, or TimelineType constants. > - **Implementation Package (`impl/`)**: Uses `package <feature>_impl`. Implements the actual tasks and imports the parent contract package. **External packages MUST NOT import the `impl` package.** > - **File Naming**: Do NOT append redundant `_task.go` or `_tasks.go` suffixes to filenames in `impl/`. Name files strictly by their DAG pipeline role (`form.go`, `query.go`, `ingester.go`, `grouper.go`, `mapper.go` or `mapper_<target>.go`, `module.go`). > - **Registration**: Tasks inside the `impl` package are declared through `impl/module.go`. --- ## 2. The Log Parsing Steps A complete log parser in KHI generally consists of distinct DAG tasks: ```mermaid flowchart TD FormTask[1. Form Task] -->|Provides Parameters| QueryTask[2. Log Query Task] QueryTask -->|Provides Raw Logs| IngesterTask[3. Log Ingest Task] QueryTask -->|Provides Raw Logs| GrouperTask[Log Grouper Task] IngesterTask -->|Provides Ingested Logs| MapperTask[4. Timeline Mapper Task] GrouperTask -->|Provides Grouped Map| MapperTask ``` ### Step 1: Form Tasks (Form-related) Exposes interactive input fields (e.g., text boxes, multi-select checkboxes) to let users configure parameters before running the inspection. - **Utility:** `formtask.DefineTextForm` or `formtask.DefineSetForm`. ### Step 2: Log Query Tasks Queries logs from the data source (e.g., Google Cloud Logging or local files) using parameters provided by the Form tasks. - **Utility:** `gcpcommon.DefineStructuredListLogEntriesTask` (for any logs on Cloud Logging) or `inspectiontaskbase.DefineInspectionTask`. - **Google Cloud API Calling:** When calling Google Cloud APIs directly or through fetchers, refer to [googlecloud-api](skill://googlecloud-api) for mandatory `CallOptionInjector` usage and client configuration. ### Step 3: Log Ingestion Tasks Extracts information directly from the log's `NodeReader` using Extractor functions, populating basic log metadata on `LogChangeSet` (such as `Timestamp` from `l.Timestamp`, `Severity`, `LogType`, and `Summary`). - **Utility:** `inspectiontaskbase.DefineLogIngesterTask`. ### Step 4: Log Grouping & Timeline Mapping Tasks - **Log Grouper Task:** Groups logs by a key (e.g., entity name, correlation ID) by calling Extractor functions on raw logs. - **Utility:** `inspectiontaskbase.DefineLogGrouperTask`. - **Timeline Mapping Task:** Maps the grouped logs to resource timelines as events or state revisions. - **Utility:** `inspectiontaskbase.DefineLogToTimelineMapperTask`. --- ## 3. Step-by-Step Implementation Code Samples Let's look at a concrete example of supporting a custom log type called `customapp` under `pkg/task/inspection/googlecloud/customapp/`. ### A. The Contract Package (`pkg/task/inspection/googlecloud/customapp/`) #### `taskid.go` Defines the TaskIDs and TaskReferences for the pipeline steps. ```go package customapp import ( inspectiontaskbase "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/taskbase" "github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid" "github.com/GoogleCloudPlatform/khi/pkg/model/log" ) const TaskIDPrefix = "customapp.khi.google.com/" // 1. Form Task ID var InputFilterKeywordTaskID = taskid.NewDefaultImplementationID[string](TaskIDPrefix + "input-keyword") // 2. Log Query Task ID var LogQueryTaskID = taskid.NewDefaultImplementationID[[]*log.Log](TaskIDPrefix + "query") // 3. Log Ingestion Task ID var LogIngesterTaskID = taskid.NewDefaultImplementationID[struct{}](TaskIDPrefix + "log-ingester") // 4. Log Grouper & Timeline Mapper Task IDs var LogGrouperTaskID = taskid.NewDefaultImplementationID[inspectiontaskbase.LogGroupMap](TaskIDPrefix + "log-grouper") var LogToTimelineMapperTaskID = taskid.NewDefaultImplementationID[struct{}](TaskIDPrefix + "timeline-mapper") ``` #### `extractor.go` Defines the strongly-typed data structures and extraction functions. > [!IMPORTANT] > > - **Package-level FieldPath declarations**: Pre-compiled `structured.FieldPath` values created by `structured.CompileFieldPath` are constant across log entries and MUST be declared as package-level variables in a `var (...)` block immediately below the `import` block. Never compile `FieldPath` inside functions or hot parsing loops. > - **Non-pointer Return Values**: Extraction methods (`ExtractXXX`) MUST return value types (`FieldSet`), not pointers (`*FieldSet`). Returning values eliminates heap allocation overhead when extractors are called millions of times across high-volume log streams. > - **Mock Support**: Extraction functions MUST check `structured.GetMock[FieldSetType](reader)` at the top of the function to allow unit tests to override extraction via `testlog.NewMockLog` / `structured.NewMockNode`. ##### Pattern 1: Direct Extractor Pattern (Single Log Source) Used when the log format is fixed to a single ingest format (e.g., GKE Autoscaler, serial port, K8s control plane). ```go package customapp import ( "github.com/GoogleCloudPlatform/khi/pkg/common/structured" ) var ( pathAppName = structured.CompileFieldPath("app_name") pathRequestID = structured.CompileFieldPath("request_id") pathPayload = structured.CompileFieldPath("payload") ) // CustomAppFieldSet holds structured log data extracted from the log entry. type CustomAppFieldSet struct { AppName string RequestID string Payload string } // ExtractCustomApp extracts CustomAppFieldSet from a raw log node reader. func ExtractCustomApp(reader *structured.NodeReader) (CustomAppFieldSet, error) { if mock, ok := structured.GetMock[CustomAppFieldSet](reader); ok { return mock, nil } return CustomAppFieldSet{ AppName: reader.ReadStringOrDefault(pathAppName, "unknown-app"), RequestID: reader.ReadStringOrDefault(pathRequestID, ""), Payload: reader.ReadStringOrDefault(pathPayload, ""), }, nil } ``` ##### Pattern 2: Injected Extractor Pattern (Multi-source Ingestion) Used when the same log entity can originate from different sources with distinct field layouts (for example, K8s audit logs ingested from GCP Cloud Logging vs. OSS Kubernetes JSONL files). The common root package defines an extractor function type and a wrapper function that accepts the extractor bound by the calling task via `coretask.Use(b, k8saudit.K8sAuditLogExtractorRef)`: ```go package k8saudit import ( "github.com/GoogleCloudPlatform/khi/pkg/common/structured" ) var K8sAuditLogCacheKey = structured.NewCacheKey[*K8sAuditLogFieldSet]() // K8sAuditLogExtractor is a function type for extracting K8sAuditLogFieldSet from a NodeReader. type K8sAuditLogExtractor func(reader *structured.NodeReader) (*K8sAuditLogFieldSet, error) // ExtractK8sAuditLog extracts K8s audit log data from a NodeReader using the given extractor. func ExtractK8sAuditLog(reader *structured.NodeReader, extractor K8sAuditLogExtractor) (*K8sAuditLogFieldSet, error) { if mock, ok := structured.GetMock[*K8sAuditLogFieldSet](reader); ok { return mock, nil } if cached, ok := structured.GetCache(reader, K8sAuditLogCacheKey); ok { return cached, nil } res, err := extractor(reader) if err == nil && res != nil { structured.SetCache(reader, K8sAuditLogCacheKey, res) } return res, err } ``` #### `timeline_type.go` Defines custom timeline types and resource verbs. ```go package customapp import ( "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6/style" ) var ( // TimelineTypeCustomApp is the timeline type style for Custom App resources. TimelineTypeCustomApp = style.MustRegisterTimelineType( "customapp", "Custom Application", "dns", 0.6, style.ColorWhite, style.ColorBlack, style.MustForceConvertSRGBHex("#4285F4"), true, 1000, style.AlphabeticalSortPolicy(), ) // VerbCustomAppProcess is the verb style for Custom App state updates. VerbCustomAppProcess = style.MustRegisterVerb("Process", style.MustForceConvertSRGBHex("#0F9D58"), style.ColorWhite, true) ) ``` #### `log_type.go` Defines custom log types. ```go package customapp import ( "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6/style" ) var ( // LogTypeCustomApp is the log type style for Custom App logs. LogTypeCustomApp = style.MustRegisterLogType( "customapp", "Custom Application Logs", style.MustForceConvertSRGBHex("#4285F4"), style.ColorWhite, ) ) ``` #### `timeline_path.go` (Optional) Defines helper functions to build hierarchical timeline paths. For custom application timelines, you can define helpers to construct paths consistently. If your custom application runs as part of a Kubernetes Pod, you can build a sub-timeline path nested directly under the standard Kubernetes Pod timeline by referencing standard K8s timeline types from `inspectioncore`. - MustXXXTimeline func must receive the context as its first argument. - If the MustXXXTimeline func isn't for a root timeline, it must receive the parent timeline path as its second argument. ```go package customapp import ( "context" "github.com/GoogleCloudPlatform/khi/pkg/common/khictx" khifilev6 "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore" ) // MustCustomAppTimeline returns the hierarchical timeline path for a standalone Custom App. // Constructs a path like: customapp/<appName> func MustCustomAppTimeline(ctx context.Context, appName string) *khifilev6.TimelinePath { builder := khictx.MustGetValue(ctx, inspectioncore.Builder) return builder.TimelineAccumulator.GetPath(nil, khifilev6.PathSegment{ Name: appName, Type: TimelineTypeCustomApp, }) } // MustCustomAppPodTimeline returns the hierarchical timeline path for Custom App logs nested under a Pod. // Constructs a path like: <apiVersion>/<kind>/<namespace>/<podName>/customapp func MustCustomAppPodTimeline(ctx context.Context, podTimelinePath *khifilev6.TimelinePath) *khifilev6.TimelinePath { if podTimelinePath == nil || podTimelinePath.Type.GetId() != inspectioncore.TimelineTypeResource.GetId() { panic("parent timeline path must be Resource type") } builder := khictx.MustGetValue(ctx, inspectioncore.Builder) return builder.TimelineAccumulator.GetPath(podTimelinePath, khifilev6.PathSegment{ Name: "customapp", Type: TimelineTypeCustomApp, }) } ``` --- ### B. The Implementation Package (`pkg/task/inspection/googlecloud/customapp/impl/`) #### `form.go` (Step 1) Implements form tasks to get user-defined input. ```go package customapp_impl import ( "context" "github.com/GoogleCloudPlatform/khi/pkg/core/inspection/formtask" coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloud/customapp" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloud/gcpcommon" ) const formPriority = gcpcommon.FormBasePriority + 5000 // InputFilterKeywordTask defines a text input form task for filtering logs. var InputFilterKeywordTask = formtask.DefineTextForm( customapp.InputFilterKeywordTaskID, formPriority, "Filter Keyword", "Keyword to filter Custom App logs.", func(b *coretask.Binder) formtask.TextFormSpec[string] { return formtask.TextFormSpec[string]{ DefaultValue: formtask.PreviousOrDefaultValue(func(ctx context.Context) (string, error) { return "default-keyword", nil }), } }, ) ``` #### `query.go` (Step 2) Implements querying logs from Google Cloud Logging based on parameters. The bind function declares each input with `coretask.Use` and passes the handles to the query source, which reads them with `Get` while the task runs. ```go package customapp_impl import ( "context" "fmt" "github.com/GoogleCloudPlatform/khi/pkg/api/googlecloud/logestimator" coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloud/customapp" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloud/gcpcommon" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloud/k8scommon" "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore" ) // LogQueryTask executes Cloud Logging queries to fetch logs. var LogQueryTask = gcpcommon.DefineStructuredListLogEntriesTask( customapp.LogQueryTaskID, "Custom App logs", func(b *coretask.Binder) gcpcommon.StructuredLogQuerySource { return &customAppQuerySource{ cluster: coretask.Use(b, k8scommon.ClusterIdentityTaskID.Ref()), keyword: coretask.Use(b, customapp.InputFilterKeywordTaskID.Ref()), } }, ) // customAppQuerySource builds the Cloud Logging query for Custom App logs from the cluster identity and the filter keyword. type customAppQuerySource struct { cluster coretask.Input[k8scommon.GoogleCloudClusterIdentity] keyword coretask.Input[string] } var _ gcpcommon.StructuredLogQuerySource = (*customAppQuerySource)(nil) func (s *customAppQuerySource) Queries(ctx context.Context, taskMode inspectioncore.InspectionTaskModeType) ([]*logestimator.StructuredLogQuery, error) { cluster := s.cluster.Get(ctx) keyword := s.keyword.Get(ctx) return []*logestimator.StructuredLogQuery{{ Incomplete: !cluster.IsComplete(), ResourceTypes: []string{"k8s_container"}, Filters: []logestimator.LoggingMonitoringMatcher{ logestimator.ResourceLabel("project_id", logestimator.Exact(cluster.ProjectID)), logestimator.LogID(logestimator.Exact("custom-app")),