From 020ac8c313c96a694af4325b153a493380e3fbf2 Mon Sep 17 00:00:00 2001 From: Michaela Lang Date: Sun, 21 Jun 2026 18:05:05 +0200 Subject: [PATCH 1/2] Refactor: Extract getLogRequestParams and add time-range filtering to Splunk queries --- pkg/api/server/v1alpha2/plugin/plugin_logs.go | 68 +++++++++---------- 1 file changed, 31 insertions(+), 37 deletions(-) diff --git a/pkg/api/server/v1alpha2/plugin/plugin_logs.go b/pkg/api/server/v1alpha2/plugin/plugin_logs.go index a1f39e0a86..99f9115161 100644 --- a/pkg/api/server/v1alpha2/plugin/plugin_logs.go +++ b/pkg/api/server/v1alpha2/plugin/plugin_logs.go @@ -119,37 +119,23 @@ func (s *LogServer) GetLog(req *pb3.GetLogRequest, srv pb3.Logs_GetLogServer) er return nil } -func getLokiLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record) error { - URL, err := url.Parse(s.config.LOGGING_PLUGIN_API_URL) - if err != nil { - s.logger.Error(err) - return err - } - URL.Path = path.Join(URL.Path, s.config.LOGGING_PLUGIN_PROXY_PATH, lokiQueryPath) - - var startTime, endTime, uidKey string +func (s *LogServer) getLogRequestParams(rec *db.Record) (startTime, endTime, uidKey string, err error) { switch rec.Type { case typePipelineRun: uidKey = pipelineRunUIDKey data := &pipelinev1.PipelineRun{} err := json.Unmarshal(rec.Data, data) if err != nil { - err = fmt.Errorf("failed to marshal pipelinerun data for fetching log, err: %s", err.Error()) - s.logger.Error(err) - return err + return "", "", "", fmt.Errorf("failed to marshal pipelinerun data for fetching log, err: %w", err) } if data.Status.StartTime == nil { - err = errors.New("there's no startime in pipelinerun") - s.logger.Error(err) - return err + return "", "", "", errors.New("there's no startime in pipelinerun") } startTime = strconv.FormatInt(data.Status.StartTime.UTC().Unix(), 10) if data.Status.CompletionTime == nil { - err = errors.New("there's no completion in pipelinerun") - s.logger.Error(err) - return err + return "", "", "", errors.New("there's no completion in pipelinerun") } endTime = strconv.FormatInt(data.Status.CompletionTime.Add(s.forwarderDelayDuration).UTC().Unix(), 10) @@ -158,27 +144,37 @@ func getLokiLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record) data := &pipelinev1.TaskRun{} err := json.Unmarshal(rec.Data, data) if err != nil { - err = fmt.Errorf("failed to marshal taskrun data for fetching log, err: %s", err.Error()) - s.logger.Error(err) - return err + return "", "", "", fmt.Errorf("failed to marshal taskrun data for fetching log, err: %w", err) } if data.Status.StartTime == nil { - err = errors.New("there's no startime in taskrun") - s.logger.Error(err) - return err + return "", "", "", errors.New("there's no startime in taskrun") } startTime = strconv.FormatInt(data.Status.StartTime.UTC().Unix(), 10) if data.Status.CompletionTime == nil { - err = errors.New("there's no completion in taskrun") - s.logger.Error(err) - return err + return "", "", "", errors.New("there's no completion in taskrun") } endTime = strconv.FormatInt(data.Status.CompletionTime.Add(s.forwarderDelayDuration).UTC().Unix(), 10) default: s.logger.Errorf("record type is invalid, record ID: %v, Name: %v, result Name: %v, result ID: %v", rec.ID, rec.Name, rec.ResultName, rec.ResultID) - return errors.New("record type is invalid") + return "", "", "", errors.New("record type is invalid") + } + return startTime, endTime, uidKey, nil +} + +func getLokiLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record) error { + URL, err := url.Parse(s.config.LOGGING_PLUGIN_API_URL) + if err != nil { + s.logger.Error(err) + return err + } + URL.Path = path.Join(URL.Path, s.config.LOGGING_PLUGIN_PROXY_PATH, lokiQueryPath) + + startTime, endTime, uidKey, err := s.getLogRequestParams(rec) + if err != nil { + s.logger.Error(err) + return err } parameters := url.Values{} @@ -536,16 +532,12 @@ func getSplunkLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record } - var uidKey string - switch rec.Type { - case typePipelineRun: - uidKey = pipelineRunUIDKey - case typeTaskRun: - uidKey = taskRunUIDKey - default: - s.logger.Errorf("record type is invalid, record ID: %v, Name: %v, result Name: %v, result ID: %v, rec Type: %v", rec.ID, rec.Name, rec.ResultName, rec.ResultID, rec.Type) - return errors.New("record type is invalid") + startTime, endTime, uidKey, err := s.getLogRequestParams(rec) + if err != nil { + s.logger.Error(err) + return err } + index, ok := s.queryParams["index"] if !ok { s.logger.Errorf("index not specified in queryParams: %v\n", s.queryParams) @@ -560,6 +552,8 @@ func getSplunkLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record queryData := url.Values{} queryData.Set("search", query) + queryData.Set("earliest_time", startTime) + queryData.Set("latest_time", endTime) req, err := http.NewRequest("POST", URL.String()+splunkOutputFormat, bytes.NewReader([]byte(queryData.Encode()))) if err != nil { From 0bcdffdfdceeeea6fa52abf912711bd9ca0d4e84 Mon Sep 17 00:00:00 2001 From: Michaela Lang Date: Sun, 21 Jun 2026 18:05:50 +0200 Subject: [PATCH 2/2] Feat: Make UID keys configurable via config.Config This commit represents Part 2 of the PR split. It introduces configurable UID keys for PipelineRuns and TaskRuns in the logging plugin. Instead of relying on direct `os.Getenv()` calls mid-process, this change integrates the keys into the central `config.Config` struct using `LOGGING_PLUGIN_PIPELINERUN_UID_KEY` and `LOGGING_PLUGIN_TASKRUN_UID_KEY`. Changes include: - Adding `LOGGING_PLUGIN_PIPELINERUN_UID_KEY` and `LOGGING_PLUGIN_TASKRUN_UID_KEY` to `pkg/api/server/config/config.go`. - Updating `getLogRequestParams()` to read these keys from `s.config`, falling back to the standard defaults if not provided in the configuration. - Adding a test case to verify the configurable UID key functionality. This ensures logging configuration remains centralized, aligns with the existing `mapstructure` pattern, enables validation at startup, and prevents configuration values from changing dynamically mid-process. --- pkg/api/server/config/config.go | 2 + pkg/api/server/v1alpha2/plugin/plugin_logs.go | 10 ++- .../v1alpha2/plugin/plugin_logs_test.go | 69 +++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/pkg/api/server/config/config.go b/pkg/api/server/config/config.go index e658553c10..c73d6bd4a7 100644 --- a/pkg/api/server/config/config.go +++ b/pkg/api/server/config/config.go @@ -72,6 +72,8 @@ type Config struct { LOGGING_PLUGIN_MULTIPART_REGEX string `mapstructure:"LOGGING_PLUGIN_MULTIPART_REGEX"` LOGGING_PLUGIN_JSON_MAP string `mapstructure:"LOGGING_PLUGIN_JSON_MAP"` LOGGING_PLUGIN_LINE_FORMAT string `mapstructure:"LOGGING_PLUGIN_LINE_FORMAT"` + LOGGING_PLUGIN_PIPELINERUN_UID_KEY string `mapstructure:"LOGGING_PLUGIN_PIPELINERUN_UID_KEY"` + LOGGING_PLUGIN_TASKRUN_UID_KEY string `mapstructure:"LOGGING_PLUGIN_TASKRUN_UID_KEY"` } func Get() *Config { diff --git a/pkg/api/server/v1alpha2/plugin/plugin_logs.go b/pkg/api/server/v1alpha2/plugin/plugin_logs.go index 99f9115161..03712d75d3 100644 --- a/pkg/api/server/v1alpha2/plugin/plugin_logs.go +++ b/pkg/api/server/v1alpha2/plugin/plugin_logs.go @@ -122,7 +122,10 @@ func (s *LogServer) GetLog(req *pb3.GetLogRequest, srv pb3.Logs_GetLogServer) er func (s *LogServer) getLogRequestParams(rec *db.Record) (startTime, endTime, uidKey string, err error) { switch rec.Type { case typePipelineRun: - uidKey = pipelineRunUIDKey + uidKey = s.config.LOGGING_PLUGIN_PIPELINERUN_UID_KEY + if uidKey == "" { + uidKey = pipelineRunUIDKey + } data := &pipelinev1.PipelineRun{} err := json.Unmarshal(rec.Data, data) if err != nil { @@ -140,7 +143,10 @@ func (s *LogServer) getLogRequestParams(rec *db.Record) (startTime, endTime, uid endTime = strconv.FormatInt(data.Status.CompletionTime.Add(s.forwarderDelayDuration).UTC().Unix(), 10) case typeTaskRun: - uidKey = taskRunUIDKey + uidKey = s.config.LOGGING_PLUGIN_TASKRUN_UID_KEY + if uidKey == "" { + uidKey = taskRunUIDKey + } data := &pipelinev1.TaskRun{} err := json.Unmarshal(rec.Data, data) if err != nil { diff --git a/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go b/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go index a44a520f27..f23768f118 100644 --- a/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go +++ b/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go @@ -171,6 +171,75 @@ func TestLogPluginServer_GetLog(t *testing.T) { } +func TestLogPluginServer_GetLog_WithConfigurableUIDKey(t *testing.T) { + // Create a mock Loki server + mockLoki := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Verify that the custom UID key is used in the query + if !strings.Contains(r.URL.String(), "custom_pipelinerun_uid_key") { + t.Errorf("expected query to contain custom_pipelinerun_uid_key, got %s", r.URL.String()) + } + + response := map[string]interface{}{ + "status": "success", + "data": map[string]interface{}{ + "result": []map[string]interface{}{ + { + "stream": map[string]string{}, + "values": [][]string{ + {"1625081600000000000", "Log Message 0"}, + }, + }, + }, + }, + } + json.NewEncoder(w).Encode(response) + })) + defer mockLoki.Close() + + tokenDir := t.TempDir() + tokenPath := filepath.Join(tokenDir, "token") + os.WriteFile(tokenPath, []byte("dummytoken"), 0600) + + srv, _ := server.New(&config.Config{ + LOGS_API: true, + LOGS_TYPE: "Loki", + DB_ENABLE_AUTO_MIGRATION: true, + LOGGING_PLUGIN_TOKEN_PATH: tokenPath, + LOGGING_PLUGIN_API_URL: mockLoki.URL, + LOGGING_PLUGIN_TLS_VERIFICATION_DISABLE: true, + LOGGING_PLUGIN_PIPELINERUN_UID_KEY: "custom_pipelinerun_uid_key", + }, logger.Get("info"), test.NewDB(t)) + + ctx := context.Background() + mockServer := &mockGetLogServer{ctx: ctx} + + res, _ := srv.CreateResult(ctx, &pb.CreateResultRequest{ + Parent: "foo", + Result: &pb.Result{Name: "foo/results/bar"}, + }) + + srv.CreateRecord(ctx, &pb.CreateRecordRequest{ + Parent: res.GetName(), + Record: &pb.Record{ + Name: record.FormatName(res.GetName(), "baz"), + Data: &pb.Any{ + Type: "tekton.dev/v1.PipelineRun", + Value: jsonutil.AnyBytes(t, pipelinev1.PipelineRun{ + Status: pipelinev1.PipelineRunStatus{ + PipelineRunStatusFields: pipelinev1.PipelineRunStatusFields{ + StartTime: &metav1.Time{Time: time.Now()}, + CompletionTime: &metav1.Time{Time: time.Now()}, + }, + }, + }), + }, + }, + }) + + req := &pb3.GetLogRequest{Name: log.FormatName(res.GetName(), "baz")} + srv.LogPluginServer.GetLog(req, mockServer) +} + func TestMergeLogParts(t *testing.T) { tests := []struct { name string