feat(jira): add JIRA_SKIP_UNPARSEABLE_ISSUES to continue collectIssues on parse failure (#8949)

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"

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(`<html>bad gateway maybe</html>`))
+		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(`<html>bad gateway maybe</html>`))
+		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(`<html>bad gateway maybe</html>`))
+		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'")
+		}
+	})
+}