From 54d0e8e793a129ed6388b94059cd4b9189516752 Mon Sep 17 00:00:00 2001 From: Ankur Kotwal Date: Wed, 12 Aug 2026 14:13:32 -0700 Subject: [PATCH] feat(jira): add JIRA_SKIP_UNPARSEABLE_ISSUES to continue collectIssues on parse failure (#8949) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On large Jira instances, jira:collectIssues occasionally receives an otherwise-successful (2xx) response whose body cannot be decoded as JSON — typically a truncated body from a dropped connection or, in rare misconfigured-proxy cases, an HTML body served with a 200. The default ResponseParser returns an error, which aborts the entire subtask and kills the whole Jira collection even though thousands of other pages succeeded. This is distinct from HTTP error statuses, which are already handled by the collector's retry logic; the request is specifically about parse failures on successful responses. Introduce a JIRA_SKIP_UNPARSEABLE_ISSUES env var: - default (unset or anything other than "true"): existing behaviour — surface the parse error and fail the subtask. - "true" (case-insensitive): log a warning and skip the unparseable page so the rest of the collection completes. The warning includes body length and a 256-byte prefix so operators can still investigate. Share the parser between the V2 (Server) and V3 (Cloud) collectors via a new parseIssuesResponse(logger) helper. Tests cover the happy path, the default error path, the skip-flag success path, case-insensitivity, and an explicit "false" value. --- backend/plugins/jira/tasks/issue_collector.go | 85 ++++++++++------ .../jira/tasks/issue_collector_test.go | 97 +++++++++++++++++++ 2 files changed, 150 insertions(+), 32 deletions(-) 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'") + } + }) +}