log-timeline-mapper

# KHI Log Ingestion & Timeline Mapping Guidelines This guide outlines the patterns, best practices, and testing methodologies for implementing log ingesters and timeline mappers in KHI. --- ## 1. LogIngesterFunc & DefineLogIngesterTask `LogIngesterFunc` is responsible for parsing raw logs and ingesting basic log metadata (summary, timestamp, severity, log type) into the KHI format. ### Function Signature ```go // LogIngesterFunc converts a log into the change set of its metadata, such as summary, severity, and timestamp. // It returns a nil change set to skip the log. type LogIngesterFunc = func(ctx context.Context, l *log.Log) (*khifilev6.LogChangeSet, error) ``` ### Key Implementation Guide > [!IMPORTANT] > **ChangeSet Metadata Ingestion:** `LogChangeSet` does NOT automatically fill metadata defaults. You MUST explicitly populate metadata on `LogChangeSet` (such as setting timestamp from `l.Timestamp`, severity, summary, and log type) in your `LogIngesterFunc` implementation. > > **Skipping Logs:** If the function returns `(nil, nil)`, KHI will treat this log as skipped (ignored) without producing any errors. #### Implementer Example ```go type MyLogIngester struct{} // ProcessLog parses a raw log entry and manually populates the LogChangeSet. func (i *MyLogIngester) ProcessLog(ctx context.Context, l *log.Log) (*khifilev6.LogChangeSet, error) { // 1. Create a new change set. cs, err := khifilev6.NewLogChangeSet(l) if err != nil { return nil, err } // 2. Set timestamp directly from l.Timestamp. cs.SetTimestamp(l.Timestamp) // 3. Set severity, summary, etc. using extractor or pre-registered styles. cs.SetSeverity(mySeverityStyle) cs.SetSummary(mySummaryString) return cs, nil } ``` ### Defining and Registering the LogIngester Task > [!IMPORTANT] > **Package Boundaries:** > > - **TaskID** definitions (e.g., `LogIngesterTaskID`) MUST be defined in the feature root package (`pkg/task/inspection/<provider>/<feature>`). > - **Task Implementation** definitions (e.g., `DefineLogIngesterTask`) MUST be placed in the `impl` subpackage (`pkg/task/inspection/<provider>/<feature>/impl`). ```go // Defined in feature root package (e.g. package myfeature): var MyLogIngesterTaskID = taskid.NewDefaultImplementationID[struct{}]("my-log-ingester") // Defined in 'impl' package (e.g. package myfeature_impl): var MyLogIngesterTask = inspectiontaskbase.DefineLogIngesterTask( myfeature.MyLogIngesterTaskID, myfeature.RawLogTaskID.Ref(), func(b *coretask.Binder) inspectiontaskbase.LogIngesterFunc { ingester := &MyLogIngester{} return ingester.ProcessLog }, ) // Register task to core runner... ``` --- ## 2. TimelineMapper & DefineLogToTimelineMapperTask `TimelineMapper` maps grouped logs to timeline elements (events or resource revisions). Depending on the complexity, you should choose one of the following three implementation patterns. ### Pattern 1: Multi-Pass with State Used for complex scenarios where you need to pre-collect information across all logs in a group before applying timeline changes (e.g., matching asynchronous request/response cycles). - **How to implement:** Implement the full `TimelineMapper[T]` interface manually. ```go type ComplexMapper struct{} func (m *ComplexMapper) PassCount() int { return 1 // Run 1 pre-processing pass. } func (m *ComplexMapper) PreProcessLogByGroup(ctx context.Context, passIndex int, l *log.Log, prevGroupData MyState) (MyState, error) { // Pre-collect state from logs. nextState := analyzeLog(prevGroupData, l) return nextState, nil } func (m *ComplexMapper) ProcessLogByGroup(ctx context.Context, l *log.Log, prevGroupData MyState) (*khifilev6.TimelineChangeSet, MyState, error) { // 1. Retrieve field data using extractor function. customSet, err := myfeature.ExtractCustom(l.NodeReader) if err != nil { return nil, prevGroupData, err } // 2. Retrieve the Builder from context. builder := khictx.MustGetValue(ctx, inspectioncore.Builder) // 3. Resolve target path dynamically using the accumulator facade. targetPath := builder.TimelineAccumulator.GetPath(nil, khifilev6.PathSegment{ Name: "complex-timeline", Type: myfeature.TimelineTypeComplex, // Timeline styles should be imported from feature root package }) cs := khifilev6.NewTimelineChangeSet(l) // Add a revision or event conditionally using the pre-collected state and customSet fields. if prevGroupData.ShouldRegisterRevision(l) { cs.AddRevision(targetPath, &khifilev6.StagingRevision{ ChangedTime: l.Timestamp, ResourceBody: customSet.Body, Principal: customSet.Principal, VerbType: myfeature.VerbCreate, }) } return cs, prevGroupData, nil } // Explicit interface compliance assertion. var _ inspectiontaskbase.TimelineMapper[MyState] = (*ComplexMapper)(nil) ``` --- ### Pattern 2: Single-Pass with State Used when you need to maintain and propagate state sequentially through the logs in a group, but do not require a pre-processing pass. - **How to implement:** Embed `SinglePassMapperBase[T]` into your mapper structure. This automatically implements `PassCount() int` (returning 0) and `PreProcessLogByGroup` (returning state as-is). ```go type StateTrackingMapper struct { inspectiontaskbase.SinglePassMapperBase[MyState] // Embeds single pass helper. } func (m *StateTrackingMapper) ProcessLogByGroup(ctx context.Context, l *log.Log, prevGroupData MyState) (*khifilev6.TimelineChangeSet, MyState, error) { // 1. Retrieve field data using extractor function. customSet, err := myfeature.ExtractCustom(l.NodeReader) if err != nil { return nil, prevGroupData, err } // 2. Maintain state. nextState := updateState(prevGroupData, customSet) // 3. Retrieve the Builder from context. builder := khictx.MustGetValue(ctx, inspectioncore.Builder) // 4. Resolve target path dynamically using the accumulator facade. targetPath := builder.TimelineAccumulator.GetPath(nil, khifilev6.PathSegment{ Name: "stateful-revision-timeline", Type: myfeature.TimelineTypeStateful, }) cs := khifilev6.NewTimelineChangeSet(l) // Append resource revision history sequentially to the timeline. cs.AddRevision(targetPath, &khifilev6.StagingRevision{ ChangedTime: l.Timestamp, ResourceBody: customSet.Body, Principal: customSet.Principal, VerbType: myfeature.VerbUpdate, }) return cs, nextState, nil } // Explicit interface compliance assertion. var _ inspectiontaskbase.TimelineMapper[MyState] = (*StateTrackingMapper)(nil) ``` --- ### Pattern 3: Single-Pass Stateless (Most Common) Used when timeline mapping for each log is completely independent and does not rely on other logs in the same group. - **How to implement:** Embed `StatelessMapperBase` into your mapper structure. This binds the state type `T` to `struct{}` and implements the pre-processing methods as no-ops. ```go type SimpleEventMapper struct { inspectiontaskbase.StatelessMapperBase // Embeds stateless helper. } func (m *SimpleEventMapper) ProcessLogByGroup(ctx context.Context, l *log.Log, _ struct{}) (*khifilev6.TimelineChangeSet, struct{}, error) { // 1. Retrieve the Builder from context. builder := khictx.MustGetValue(ctx, inspectioncore.Builder) // 2. Resolve target path dynamically. targetPath := builder.TimelineAccumulator.GetPath(nil, khifilev6.PathSegment{ Name: "simple-event-timeline", Type: myfeature.TimelineTypeEvent, }) cs := khifilev6.NewTimelineChangeSet(l) // Add a simple timeline event. cs.AddEvent(targetPath) return cs, struct{}{}, nil } // Explicit interface compliance assertion. var _ inspectiontaskbase.TimelineMapper[struct{}] = (*SimpleEventMapper)(nil) ``` ### Defining and Registering the TimelineMapper Task > [!IMPORTANT] > **Package Boundaries:** > > - **TaskID** definitions (e.g., `LogToTimelineMapperTaskID`) MUST be defined in the feature root package. > - **Task Implementation** definitions (e.g., `DefineLogToTimelineMapperTask`) MUST be placed in the `impl` subpackage. ```go // Defined in feature root package: var MyTimelineMapperTaskID = taskid.NewDefaultImplementationID[struct{}]("my-timeline-mapper") // Defined in 'impl' package: var MyTimelineMapperTask = inspectiontaskbase.DefineLogToTimelineMapperTask( myfeature.MyTimelineMapperTaskID, inspectiontaskbase.TimelineMapperInputs{ LogIngester: myfeature.MyLogIngesterTaskID.Ref(), GroupedLogs: myfeature.MyLogGrouperTaskID.Ref(), }, func(b *coretask.Binder) inspectiontaskbase.TimelineMapper[struct{}] { return &SimpleEventMapper{} }, ) // Register task to core runner... ``` --- ## 3. Testing Guidelines Tests for tasks must follow the standard Table-Driven testing pattern. To verify mappers or ingesters produce correct outcomes across various scenarios, write unit tests utilizing `testchangeset` fluent assertions. ### Table-Driven Fluent Assertions KHI provides a dedicated test utility `github.com/GoogleCloudPlatform/khi/pkg/testutil/testchangeset` to perform readable assertions against staged changesets. By incorporating `testchangeset.AssertLog` or `testchangeset.AssertTimeline` into your table-driven loop, you can verify multiple test cases cleanly and expressively. #### LogIngester Table-Driven Assertion Example To isolate ingester parsing logic, instantiate logs using `testlog.NewMockLog` with typed field sets (which can receive multiple fieldsets and an optional `time.Time`): ```go func TestMyLogIngester_ProcessLog(t *testing.T) { testCases := []struct { name string input *log.Log assert func(t *testing.T, cs *khifilev6.LogChangeSet) }{ { name: "successful info log ingestion", input: testlog.NewMockLog( time.Date(2026, 5, 22, 12, 0, 0, 0, time.UTC), MyLogFieldSet{ Message: "server started", }, ), assert: func(t *testing.T, cs *khifilev6.LogChangeSet) { testchangeset.AssertLog(t, cs). HasSummary("server started"). HasSeverity(infoSeverityStyle) }, }, } ingester := &MyLogIngester{} for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { ctx := t.Context() cs, err := ingester.ProcessLog(ctx, tc.input) if err != nil { t.Fatalf("ProcessLog() returned unexpected error: %v", err) } tc.assert(t, cs) }) } } ``` #### TimelineMapper Table-Driven Assertion Example Mappers translate structured logs into timeline changes. When a mapper reads upstream task results via `coretask.Input[T]`, extract the per-log mapping logic into a pure helper function (e.g., `mapMyLog(ctx, l, prevState, depValue)`) so `ProcessLogByGroup` simply resolves `m.dep.Get(ctx)` and delegates to `mapMyLog`. Unit tests can then call `mapMyLog` directly with `testlog.NewMockLog` and `testchangeset.AssertTimeline` without needing a `Binder` or task harness. > [!IMPORTANT] > **Shared Builder Reference:** When unit testing mappers that dynamically construct timeline paths via context builder, you MUST initialize a single `khifilev6.Builder` (or `khifilev6.NewTestBuilder`) and resolve all comparison `TimelinePath` instances using this builder. Crucially, the same builder instance must be injected into the execution context using `khictx.WithValue` to ensure pointer equality during assertions. ```go func TestMapMyLog(t *testing.T) { // 1. Initialize the Builder first. builder := khifilev6.NewTestBuilder(id.NewGenerator()) // 2. Resolve comparative path instances using the Builder's accumulator. // TimelineTypes must be imported from the feature root package. resourceTimelinePath := builder.TimelineAccumulator.GetPath(nil, khifilev6.PathSegment{ Name: "resource-timeline", Type: myfeature.TimelineTypeResource, }) testCases := []struct { name string inputLog *log.Log prevState MyState assert func(t *testing.T, cs *khifilev6.TimelineChangeSet) }{ { name: "create resource revision", inputLog: testlog.NewMockLog( time.Date(2026, 5, 22, 12, 0, 0, 0, time.UTC), map[string]any{ "verb": "create", }, ), prevState: MyState{}, assert: func(t *testing.T, cs *khifilev6.TimelineChangeSet) { testchangeset.AssertTimeline(t, cs). HasEvent(resourceTimelinePath). HasRevision(resourceTimelinePath, &khifilev6.StagingRevision{ VerbType: myfeature.VerbCreate, }) }, }, { name: "skip timeline revision on delete verb", inputLog: testlog.NewMockLog( time.Date(2026, 5, 22, 12, 0, 0, 0, time.UTC), map[string]any{ "verb": "delete", }, ), prevState: MyState{}, assert: func(t *testing.T, cs *khifilev6.TimelineChangeSet) { testchangeset.AssertTimeline(t, cs). HasNoEvent(resourceTimelinePath). HasNoRevision(resourceTimelinePath) }, }, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { // 3. Set up the context using t.Context() and SAME builder instance. ctx := khictx.WithValue(t.Context(), inspectioncore.Builder, builder) cs, _, err := mapMyLog(ctx, tc.inputLog, tc.prevState) if err != nil { t.Fatalf("mapMyLog() returned unexpected error: %v", err) } tc.assert(t, cs) }) } } ``` To also verify task-level `Binder` wiring and end-to-end execution of `MyTimelineMapperTask`, use `inspectiontest.Run` with `tasktest.Given`: ```go _, err := inspectiontest.Run( t, ctx, MyTimelineMapperTask, inspectioncore.TaskModeRun, nil, tasktest.Given(myfeature.MyLogIngesterTaskID, struct{}{}), tasktest.Given(myfeature.MyLogGrouperTaskID, groupedLogs), ) ```