diff --git a/backend/plugins/jira/tasks/issue_collector.go b/backend/plugins/jira/tasks/issue_collector.go index 28aa473ff80..6f0caa43694 100644 --- a/backend/plugins/jira/tasks/issue_collector.go +++ b/backend/plugins/jira/tasks/issue_collector.go @@ -23,16 +23,27 @@ import ( "io" "net/http" "net/url" + "os" "strings" "time" "github.com/apache/incubator-devlake/core/dal" "github.com/apache/incubator-devlake/core/errors" + "github.com/apache/incubator-devlake/core/log" "github.com/apache/incubator-devlake/core/plugin" "github.com/apache/incubator-devlake/helpers/pluginhelper/api" "github.com/apache/incubator-devlake/plugins/jira/models" ) +// skipUnparseableIssuesEnvVar controls whether jira:collectIssues aborts on a +// JSON parse failure of an otherwise-successful (2xx) response body (default), +// or logs a warning and skips the bad page so the rest of the sync completes. +const skipUnparseableIssuesEnvVar = "JIRA_SKIP_UNPARSEABLE_ISSUES" + +func shouldSkipUnparseableIssues() bool { + return strings.EqualFold(os.Getenv(skipUnparseableIssuesEnvVar), "true") +} + const RAW_ISSUE_TABLE = "jira_api_issues" var _ plugin.SubTaskEntryPoint = CollectIssues @@ -87,10 +98,10 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error { if strings.EqualFold(string(data.JiraServerInfo.DeploymentType), string(models.DeploymentServer)) { logger.Info("Using api/2/search for JIRA Server issue collection") - err = setupIssueV2Collector(apiCollector, data, filterJql, pageSize) + err = setupIssueV2Collector(apiCollector, data, filterJql, pageSize, logger) } else { logger.Info("Using api/3/search/jql for JIRA Cloud issue collection") - err = setupIssueV3Collector(apiCollector, data, filterJql, pageSize) + err = setupIssueV3Collector(apiCollector, data, filterJql, pageSize, logger) } if err != nil { return err @@ -114,7 +125,7 @@ func buildFilterJQL(filterId string, incrementalJql string) string { return fmt.Sprintf("filter = %s AND %s", filterId, incrementalJql) } -func setupIssueV2Collector(apiCollector *api.StatefulApiCollector, data *JiraTaskData, filterJql string, pageSize int) errors.Error { +func setupIssueV2Collector(apiCollector *api.StatefulApiCollector, data *JiraTaskData, filterJql string, pageSize int, logger log.Logger) errors.Error { return apiCollector.InitCollector(api.ApiCollectorArgs{ ApiClient: data.ApiClient, PageSize: pageSize, @@ -129,24 +140,11 @@ func setupIssueV2Collector(apiCollector *api.StatefulApiCollector, data *JiraTas }, GetTotalPages: GetTotalPagesFromResponse, Concurrency: 10, - ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) { - var data struct { - Issues []json.RawMessage `json:"issues"` - } - blob, err := io.ReadAll(res.Body) - if err != nil { - return nil, errors.Convert(err) - } - err = json.Unmarshal(blob, &data) - if err != nil { - return nil, errors.Convert(err) - } - return data.Issues, nil - }, + ResponseParser: parseIssuesResponse(logger), }) } -func setupIssueV3Collector(apiCollector *api.StatefulApiCollector, data *JiraTaskData, filterJql string, pageSize int) errors.Error { +func setupIssueV3Collector(apiCollector *api.StatefulApiCollector, data *JiraTaskData, filterJql string, pageSize int, logger log.Logger) errors.Error { return apiCollector.InitCollector(api.ApiCollectorArgs{ ApiClient: data.ApiClient, PageSize: pageSize, @@ -163,23 +161,46 @@ func setupIssueV3Collector(apiCollector *api.StatefulApiCollector, data *JiraTas } return query, nil }, - ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) { - var data struct { - Issues []json.RawMessage `json:"issues"` - } - blob, err := io.ReadAll(res.Body) - if err != nil { - return nil, errors.Convert(err) - } - err = json.Unmarshal(blob, &data) - if err != nil { - return nil, errors.Convert(err) - } - return data.Issues, nil - }, + ResponseParser: parseIssuesResponse(logger), }) } +// parseIssuesResponse builds the ResponseParser shared by the V2 (Server) and V3 +// (Cloud) Jira issue collectors. When the response body of a 2xx-status page +// cannot be decoded as JSON — typically a truncated body from a dropped +// connection, or a mis-served HTML page from an upstream proxy — the default is +// to surface the error and abort collectIssues. On big/flaky Jira instances a +// single bad page can kill an entire sync while thousands of other pages are +// fine, so JIRA_SKIP_UNPARSEABLE_ISSUES=true opts into a non-fatal mode: log a +// warning and skip the page. See #8949. +func parseIssuesResponse(logger log.Logger) func(res *http.Response) ([]json.RawMessage, errors.Error) { + return func(res *http.Response) ([]json.RawMessage, errors.Error) { + var body struct { + Issues []json.RawMessage `json:"issues"` + } + blob, err := io.ReadAll(res.Body) + if err != nil { + return nil, errors.Convert(err) + } + if err := json.Unmarshal(blob, &body); err != nil { + if shouldSkipUnparseableIssues() { + preview := blob + if len(preview) > 256 { + preview = preview[:256] + } + logger.Warn( + err, + "jira:collectIssues: %s=true, skipping unparseable page (body_len=%d, prefix=%q)", + skipUnparseableIssuesEnvVar, len(blob), string(preview), + ) + return []json.RawMessage{}, nil + } + return nil, errors.Convert(err) + } + return body.Issues, nil + } +} + // buildJQL build jql based on timeAfter and incremental mode func buildJQL(since time.Time, location *time.Location) string { jql := "ORDER BY created ASC" diff --git a/backend/plugins/jira/tasks/issue_collector_test.go b/backend/plugins/jira/tasks/issue_collector_test.go index 7d5bdc1c1c9..eb9b7258ca8 100644 --- a/backend/plugins/jira/tasks/issue_collector_test.go +++ b/backend/plugins/jira/tasks/issue_collector_test.go @@ -18,10 +18,37 @@ limitations under the License. package tasks import ( + "bytes" + "io" + "net/http" "testing" "time" + + "github.com/apache/incubator-devlake/core/log" ) +// noopLogger is a Logger stub used by tests. It records the last Warn call so +// tests can assert that skip-mode logged an informative warning. +type noopLogger struct { + warnCalls int + lastWarn string +} + +func (l *noopLogger) IsLevelEnabled(log.LogLevel) bool { return false } +func (l *noopLogger) Printf(string, ...interface{}) {} +func (l *noopLogger) Log(log.LogLevel, string, ...interface{}) {} +func (l *noopLogger) Debug(string, ...interface{}) {} +func (l *noopLogger) Info(string, ...interface{}) {} +func (l *noopLogger) Warn(_ error, format string, a ...interface{}) { + l.warnCalls++ + l.lastWarn = format + _ = a +} +func (l *noopLogger) Error(error, string, ...interface{}) {} +func (l *noopLogger) Nested(string) log.Logger { return l } +func (l *noopLogger) GetConfig() *log.LoggerConfig { return &log.LoggerConfig{} } +func (l *noopLogger) SetStream(*log.LoggerStreamConfig) {} + func Test_buildJQL(t *testing.T) { base := time.Date(2021, 2, 3, 4, 5, 6, 7, time.UTC) timeAfter := base @@ -103,3 +130,73 @@ func Test_buildFilterJQL(t *testing.T) { }) } } + +// Test_parseIssuesResponse verifies that the shared jira:collectIssues response +// parser: (a) still returns the parsed issues on the happy path; (b) still +// surfaces the parse error by default when JIRA_SKIP_UNPARSEABLE_ISSUES is not +// set — preserving pre-#8949 behaviour; (c) logs a warning and returns an empty +// slice + nil error when JIRA_SKIP_UNPARSEABLE_ISSUES=true. +func Test_parseIssuesResponse(t *testing.T) { + newRes := func(body string) *http.Response { + return &http.Response{Body: io.NopCloser(bytes.NewBufferString(body))} + } + + t.Run("happy path returns issues", func(t *testing.T) { + logger := &noopLogger{} + got, err := parseIssuesResponse(logger)(newRes(`{"issues":[{"id":"1"},{"id":"2"}]}`)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 2 { + t.Fatalf("want 2 issues, got %d", len(got)) + } + if logger.warnCalls != 0 { + t.Fatalf("did not expect a warn call on happy path, got %d", logger.warnCalls) + } + }) + + t.Run("default returns error on unparseable body", func(t *testing.T) { + t.Setenv(skipUnparseableIssuesEnvVar, "") + logger := &noopLogger{} + _, err := parseIssuesResponse(logger)(newRes(`bad gateway maybe`)) + if err == nil { + t.Fatal("expected an error when body is unparseable and skip flag is unset") + } + if logger.warnCalls != 0 { + t.Fatalf("did not expect a warn call in default mode, got %d", logger.warnCalls) + } + }) + + t.Run("skip flag logs warning and returns empty slice", func(t *testing.T) { + t.Setenv(skipUnparseableIssuesEnvVar, "true") + logger := &noopLogger{} + got, err := parseIssuesResponse(logger)(newRes(`bad gateway maybe`)) + if err != nil { + t.Fatalf("expected no error when skip flag is set, got: %v", err) + } + if got == nil || len(got) != 0 { + t.Fatalf("expected empty (non-nil) slice on skip, got %v", got) + } + if logger.warnCalls != 1 { + t.Fatalf("expected one warn call, got %d", logger.warnCalls) + } + }) + + t.Run("skip flag is case-insensitive", func(t *testing.T) { + t.Setenv(skipUnparseableIssuesEnvVar, "TRUE") + logger := &noopLogger{} + _, err := parseIssuesResponse(logger)(newRes(`bad gateway maybe`)) + if err != nil { + t.Fatalf("expected no error with TRUE, got: %v", err) + } + }) + + t.Run("skip flag off with 'false' surfaces error", func(t *testing.T) { + t.Setenv(skipUnparseableIssuesEnvVar, "false") + logger := &noopLogger{} + _, err := parseIssuesResponse(logger)(newRes(`{`)) + if err == nil { + t.Fatal("expected error when flag is 'false'") + } + }) +}