Skip to content

Commit 347827a

Browse files
kyasbalkyasbal
authored
fix composer parsers to use GCPMainMessageFieldSet because log.MainMessageFieldSet is no longer used by default (#792)
fix composer parsers to use GCPMainMessageFieldSet because log.MainMesageFieldSet is no longer used by default
1 parent e9d39b2 commit 347827a

8 files changed

Lines changed: 22 additions & 30 deletions

File tree

‎pkg/model/log/common_field_set.go‎

Lines changed: 0 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,3 @@ func (c *CommonFieldSet) Kind() string {
3030
}
3131

3232
var _ FieldSet = (*CommonFieldSet)(nil)
33-
34-
// MainMessageFieldSet is an abstract FieldSet struct type to get the main message of its log.
35-
// This would be read from `textPayload`, `protoPayload` or `jsonPayload` when it is read from Cloud Logging.
36-
//
37-
// Deprecated: Define custom FieldSet and FieldSetReader instead.
38-
type MainMessageFieldSet struct {
39-
MainMessage string
40-
}
41-
42-
// Kind implements FieldSet.
43-
func (d *MainMessageFieldSet) Kind() string {
44-
return "main_message"
45-
}
46-
47-
var _ FieldSet = (*MainMessageFieldSet)(nil)

‎pkg/task/inspection/googlecloudclustercomposer/impl/dag_processor_manager_mapper.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2929
commonlogk8saudit_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/commonlogk8saudit/contract"
3030
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
31+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
3132
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
3233
)
3334

@@ -104,7 +105,7 @@ func (i *dagProcessorManagerLogIngester) ProcessLogByGroup(ctx context.Context,
104105
cs.SetSeverity(inspectioncore_contract.SeverityUnknown)
105106
cs.SetSummary("")
106107

107-
mainMessage, err := log.GetFieldSet(l, &log.MainMessageFieldSet{})
108+
mainMessage, err := log.GetFieldSet(l, &googlecloudcommon_contract.GCPMainMessageFieldSet{})
108109
if err != nil {
109110
return cs, prevGroupData, nil
110111
}
@@ -183,7 +184,7 @@ func (m *dagProcessorManagerTimelineMapper) ProcessLogByGroup(ctx context.Contex
183184
envPath := googlecloudclustercomposer_contract.MustComposerEnvironmentTimeline(ctx, clusterIdentity.ProjectID, environmentName)
184185

185186
commonField, _ := log.GetFieldSet(l, &log.CommonFieldSet{})
186-
mainMessage, err := log.GetFieldSet(l, &log.MainMessageFieldSet{})
187+
mainMessage, err := log.GetFieldSet(l, &googlecloudcommon_contract.GCPMainMessageFieldSet{})
187188
if err != nil {
188189
return nil, prevGroupData, nil
189190
}

‎pkg/task/inspection/googlecloudclustercomposer/impl/dag_processor_manager_mapper_test.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
core_contract "github.com/GoogleCloudPlatform/khi/pkg/task/core/contract"
3030
commonlogk8saudit_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/commonlogk8saudit/contract"
3131
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
32+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
3233
googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
3334
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
3435
"github.com/GoogleCloudPlatform/khi/pkg/testutil/testchangeset"
@@ -44,19 +45,19 @@ func TestDagProcessorMapperTask_ProcessLogByGroup(t *testing.T) {
4445
logsCase1 := []*log.Log{
4546
log.NewLogWithFieldSetsForTest(
4647
&log.CommonFieldSet{Timestamp: timestamp2},
47-
&log.MainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: =========== DAG File Processing Stats ============"},
48+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: =========== DAG File Processing Stats ============"},
4849
),
4950
log.NewLogWithFieldSetsForTest(
5051
&log.CommonFieldSet{Timestamp: timestamp3},
51-
&log.MainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: File Path PID Runtime # DAGs # Errors Last Runtime Last Run"},
52+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: File Path PID Runtime # DAGs # Errors Last Runtime Last Run"},
5253
),
5354
log.NewLogWithFieldSetsForTest(
5455
&log.CommonFieldSet{Timestamp: timestamp4},
55-
&log.MainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: -------------------------------------------------- ----- --------- -------- ---------- -------------- -------------------"},
56+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: -------------------------------------------------- ----- --------- -------- ---------- -------------- -------------------"},
5657
),
5758
log.NewLogWithFieldSetsForTest(
5859
&log.CommonFieldSet{Timestamp: timestamp5},
59-
&log.MainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: /home/airflow/gcs/dags/airflow_monitoring.py 1 0 0.36s 2026-03-08T04:49:37"},
60+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "DAG_PROCESSOR_MANAGER_LOG: /home/airflow/gcs/dags/airflow_monitoring.py 1 0 0.36s 2026-03-08T04:49:37"},
6061
),
6162
}
6263

@@ -192,7 +193,7 @@ func TestDagProcessorLogIngester_ProcessLog(t *testing.T) {
192193
for _, msg := range tc.messages {
193194
inputLog := log.NewLogWithFieldSetsForTest(
194195
&log.CommonFieldSet{Timestamp: timestamp},
195-
&log.MainMessageFieldSet{MainMessage: msg},
196+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: msg},
196197
)
197198
cs, nextState, err := ingester.ProcessLogByGroup(context.Background(), inputLog, state)
198199
if err != nil {

‎pkg/task/inspection/googlecloudclustercomposer/impl/other.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
khifilev6 "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6"
2424
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2525
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
26+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
2627
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
2728
)
2829

@@ -73,7 +74,7 @@ func (i *otherLogIngester) ProcessLog(ctx context.Context, l *log.Log) (*khifile
7374
cs.SetSeverity(inspectioncore_contract.SeverityUnknown)
7475
}
7576

76-
if messageFS, err := log.GetFieldSet(l, &log.MainMessageFieldSet{}); err == nil {
77+
if messageFS, err := log.GetFieldSet(l, &googlecloudcommon_contract.GCPMainMessageFieldSet{}); err == nil {
7778
cs.SetSummary(messageFS.MainMessage)
7879
}
7980

‎pkg/task/inspection/googlecloudclustercomposer/impl/scheduler.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
khifilev6 "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6"
2525
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2626
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
27+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
2728
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
2829
)
2930

@@ -68,7 +69,7 @@ func (i *schedulerLogIngester) ProcessLog(ctx context.Context, l *log.Log) (*khi
6869
cs.SetSeverity(inspectioncore_contract.SeverityUnknown)
6970
}
7071

71-
if messageFS, err := log.GetFieldSet(l, &log.MainMessageFieldSet{}); err == nil {
72+
if messageFS, err := log.GetFieldSet(l, &googlecloudcommon_contract.GCPMainMessageFieldSet{}); err == nil {
7273
cs.SetSummary(messageFS.MainMessage)
7374
}
7475

‎pkg/task/inspection/googlecloudclustercomposer/impl/scheduler_test.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2828
core_contract "github.com/GoogleCloudPlatform/khi/pkg/task/core/contract"
2929
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
30+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
3031
googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
3132
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
3233
"github.com/GoogleCloudPlatform/khi/pkg/testutil/testchangeset"
@@ -45,7 +46,7 @@ func TestAirflowSchedulerMapperTask_ProcessLogByGroup(t *testing.T) {
4546
name: "Scheduler basic identification and TaskInstance extraction",
4647
input: log.NewLogWithFieldSetsForTest(
4748
&log.CommonFieldSet{Timestamp: timestamp},
48-
&log.MainMessageFieldSet{MainMessage: "Processing /app/models.py"},
49+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "Processing /app/models.py"},
4950
&googlecloudclustercomposer_contract.ComposerFieldSet{
5051
SchedulerID: "airflow-scheduler-7b5f",
5152
},
@@ -84,7 +85,7 @@ func TestAirflowSchedulerMapperTask_ProcessLogByGroup(t *testing.T) {
8485
name: "Zombie task adds event to worker",
8586
input: log.NewLogWithFieldSetsForTest(
8687
&log.CommonFieldSet{Timestamp: timestamp},
87-
&log.MainMessageFieldSet{MainMessage: "Detected zombie task"},
88+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "Detected zombie task"},
8889
&googlecloudclustercomposer_contract.ComposerFieldSet{
8990
SchedulerID: "airflow-scheduler-7b5f",
9091
},
@@ -125,7 +126,7 @@ func TestAirflowSchedulerMapperTask_ProcessLogByGroup(t *testing.T) {
125126
name: "Scheduler log without TaskInstance",
126127
input: log.NewLogWithFieldSetsForTest(
127128
&log.CommonFieldSet{Timestamp: timestamp},
128-
&log.MainMessageFieldSet{MainMessage: "Heartbeat"},
129+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "Heartbeat"},
129130
&googlecloudclustercomposer_contract.ComposerFieldSet{
130131
SchedulerID: "airflow-scheduler-7b5f",
131132
},

‎pkg/task/inspection/googlecloudclustercomposer/impl/worker.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
khifilev6 "github.com/GoogleCloudPlatform/khi/pkg/model/khifile/v6"
2525
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2626
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
27+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
2728
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
2829
)
2930

@@ -68,7 +69,7 @@ func (i *workerLogIngester) ProcessLog(ctx context.Context, l *log.Log) (*khifil
6869
cs.SetSeverity(inspectioncore_contract.SeverityUnknown)
6970
}
7071

71-
if messageFS, err := log.GetFieldSet(l, &log.MainMessageFieldSet{}); err == nil {
72+
if messageFS, err := log.GetFieldSet(l, &googlecloudcommon_contract.GCPMainMessageFieldSet{}); err == nil {
7273
cs.SetSummary(messageFS.MainMessage)
7374
}
7475

‎pkg/task/inspection/googlecloudclustercomposer/impl/worker_test.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/GoogleCloudPlatform/khi/pkg/model/log"
2828
core_contract "github.com/GoogleCloudPlatform/khi/pkg/task/core/contract"
2929
googlecloudclustercomposer_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustercomposer/contract"
30+
googlecloudcommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudcommon/contract"
3031
googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
3132
inspectioncore_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/inspectioncore/contract"
3233
"github.com/GoogleCloudPlatform/khi/pkg/testutil/testchangeset"
@@ -45,7 +46,7 @@ func TestAirflowWorkerMapperTask_ProcessLogByGroup(t *testing.T) {
4546
name: "Worker basic identification and TaskInstance extraction",
4647
input: log.NewLogWithFieldSetsForTest(
4748
&log.CommonFieldSet{Timestamp: timestamp},
48-
&log.MainMessageFieldSet{MainMessage: "Executing task"},
49+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "Executing task"},
4950
&googlecloudclustercomposer_contract.ComposerFieldSet{
5051
WorkerID: "airflow-worker-abc",
5152
},
@@ -83,7 +84,7 @@ func TestAirflowWorkerMapperTask_ProcessLogByGroup(t *testing.T) {
8384
name: "Worker log without TaskInstance",
8485
input: log.NewLogWithFieldSetsForTest(
8586
&log.CommonFieldSet{Timestamp: timestamp},
86-
&log.MainMessageFieldSet{MainMessage: "Worker Heartbeat"},
87+
&googlecloudcommon_contract.GCPMainMessageFieldSet{MainMessage: "Worker Heartbeat"},
8788
&googlecloudclustercomposer_contract.ComposerFieldSet{
8889
WorkerID: "airflow-worker-abc",
8990
},

0 commit comments

Comments
 (0)