Infer each degradation in goroutine and limit GOMAXPROCS to the kube …

…limits

cmd/degradation-detector/main.go

@@ -4,6 +4,7 @@ import (
   "context"
   detector "github.com/JetBrains/ij-perf-report-aggregator/pkg/degradation-detector"
   "github.com/JetBrains/ij-perf-report-aggregator/pkg/degradation-detector/analysis"
+  _ "go.uber.org/automaxprocs"
   "log/slog"
   "net/http"
   "os"
@@ -12,46 +13,77 @@ import (
 )
 
 func main() {
+  backendUrl := getBackendUrl()
+  client := createHttpClient()
+  analysisSettings := generateAnalysisSettings(backendUrl, client)
+  degradations := getDegradations(analysisSettings, client, backendUrl)
+  insertionResults := writeDegradations(client, backendUrl, degradations)
+  postDegradations(insertionResults, client)
+}
+
+func getBackendUrl() string {
   backendUrl := os.Getenv("BACKEND_URL")
-  if len(backendUrl) == 0 {
-    backendUrl = "https://ij-perf-api.labs.jb.gg" //http://localhost:9044
+  if backendUrl == "" {
+    backendUrl = "https://ij-perf-api.labs.jb.gg" // Default URL
     slog.Info("BACKEND_URL is not set, using default value: %s", "url", backendUrl)
   }
+  return backendUrl
+}
 
-  client := &http.Client{
+func createHttpClient() *http.Client {
+  return &http.Client{
     Timeout: 60 * time.Second,
     Transport: &http.Transport{
       MaxIdleConns:        20,
       MaxIdleConnsPerHost: 10,
     },
   }
+}
 
-  analysisSettings := make([]detector.Settings, 0, 1000)
-  analysisSettings = append(analysisSettings, analysis.GenerateIdeaSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GenerateWorkspaceSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GenerateKotlinSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GenerateMavenSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GenerateGradleSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GeneratePhpStormSettings()...)
-  analysisSettings = append(analysisSettings, analysis.GenerateUnitTestsSettings(backendUrl, client)...)
+func generateAnalysisSettings(backendUrl string, client *http.Client) []detector.Settings {
+  settings := make([]detector.Settings, 0, 1000)
+  settings = append(settings, analysis.GenerateIdeaSettings()...)
+  settings = append(settings, analysis.GenerateWorkspaceSettings()...)
+  settings = append(settings, analysis.GenerateKotlinSettings()...)
+  settings = append(settings, analysis.GenerateMavenSettings()...)
+  settings = append(settings, analysis.GenerateGradleSettings()...)
+  settings = append(settings, analysis.GeneratePhpStormSettings()...)
+  settings = append(settings, analysis.GenerateUnitTestsSettings(backendUrl, client)...)
+  return settings
+}
 
-  degradations := make([]detector.Degradation, 0, 1000)
+func getDegradations(analysisSettings []detector.Settings, client *http.Client, backendUrl string) []detector.Degradation {
+  degradationChan := make(chan []detector.Degradation)
+  var wgAnalysis sync.WaitGroup
   for _, analysisSetting := range analysisSettings {
-    slog.Info("processing", "settings", analysisSetting)
-    ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
-    timestamps, values, builds, err := detector.GetDataFromClickhouse(ctx, client, backendUrl, analysisSetting)
-    if err != nil {
-      slog.Error("error while getting data from clickhouse", "error", err)
-    }
-
-    degradations = append(degradations, detector.InferDegradations(values, builds, timestamps, analysisSetting)...)
-    cancel()
+    wgAnalysis.Add(1)
+    go func(analysisSetting detector.Settings) {
+      defer wgAnalysis.Done()
+      slog.Info("processing", "settings", analysisSetting)
+      ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+      defer cancel()
+      timestamps, values, builds, err := detector.GetDataFromClickhouse(ctx, client, backendUrl, analysisSetting)
+      if err != nil {
+        slog.Error("error while getting data from clickhouse", "error", err, "settings", analysisSetting)
+        return
+      }
+      degradationChan <- detector.InferDegradations(values, builds, timestamps, analysisSetting)
+    }(analysisSetting)
   }
 
-  insertionCtx, cancelInsertion := context.WithTimeout(context.Background(), 5*time.Minute)
-  defer cancelInsertion()
-  insertionResults := detector.PostDegradations(insertionCtx, client, backendUrl, degradations)
+  go func() {
+    wgAnalysis.Wait()
+    close(degradationChan)
+  }()
+
+  degradations := make([]detector.Degradation, 0, 1000)
+  for d := range degradationChan {
+    degradations = append(degradations, d...)
+  }
+  return degradations
+}
 
+func postDegradations(insertionResults []detector.InsertionResults, client *http.Client) {
   var wg sync.WaitGroup
   for _, result := range insertionResults {
     if result.Error != nil {
@@ -62,9 +94,9 @@ func main() {
       continue
     }
     wg.Add(1)
-    ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
     go func(result detector.InsertionResults) {
       defer wg.Done()
+      ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
       err := detector.SendSlackMessage(ctx, client, result.Degradation)
       if err != nil {
         slog.Error("error while sending slack message", "error", err)
@@ -74,3 +106,9 @@ func main() {
   }
   wg.Wait()
 }
+
+func writeDegradations(client *http.Client, backendUrl string, degradations []detector.Degradation) []detector.InsertionResults {
+  ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+  defer cancel()
+  return detector.PostDegradations(ctx, client, backendUrl, degradations)
+}

go.mod

@@ -149,6 +149,7 @@ require (
 	go.opentelemetry.io/otel v1.21.0 // indirect
 	go.opentelemetry.io/otel/metric v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.21.0 // indirect
+	go.uber.org/automaxprocs v1.5.3 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go4.org v0.0.0-20230225012048-214862532bf5 // indirect
 	golang.org/x/crypto v0.15.0 // indirect

go.sum

@@ -461,6 +461,8 @@ go.opentelemetry.io/otel/trace v1.21.0 h1:WD9i5gzvoUPuXIXH24ZNBudiarZDKuekPqi/E8
 go.opentelemetry.io/otel/trace v1.21.0/go.mod h1:LGbsEB0f9LGjN+OZaQQ26sohbOmiMR+BaslueVtS/qQ=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
+go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
+go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
 go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=