Merge pull request #54 from opengovern/feat-make-task

fix: create task for describer

Author: artaasadi

discovery/envs/envs.go

@@ -0,0 +1,24 @@
+package envs
+
+import (
+	"github.com/opengovern/opensecurity/services/tasks/worker/consts"
+	"os"
+)
+
+var (
+	NatsURL         = os.Getenv(consts.NatsURLEnv)
+	NatsConsumer    = os.Getenv(consts.NatsConsumerEnv)
+	StreamName      = os.Getenv(consts.NatsStreamNameEnv)
+	TopicName       = os.Getenv(consts.NatsTopicNameEnv)
+	ResultTopicName = os.Getenv(consts.NatsResultTopicNameEnv)
+
+	ESAddress       = os.Getenv(consts.ElasticSearchAddressEnv)
+	ESUsername      = os.Getenv(consts.ElasticSearchUsernameEnv)
+	ESPassword      = os.Getenv(consts.ElasticSearchPasswordEnv)
+	ESIsOnAks       = os.Getenv(consts.ElasticSearchIsOnAksNameEnv)
+	ESIsOpenSearch  = os.Getenv(consts.ElasticSearchIsOpenSearch)
+	ESAwsRegion     = os.Getenv(consts.ElasticSearchAwsRegionEnv)
+	ESAssumeRoleArn = os.Getenv(consts.ElasticSearchAssumeRoleArnEnv)
+
+	InventoryServiceEndpoint = os.Getenv(consts.InventoryBaseURL)
+)

discovery/main.go

@@ -3,9 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/opengovern/og-describer-github/discovery/pkg"
-	"github.com/spf13/cobra"
-	"go.uber.org/zap"
+	"github.com/opengovern/og-describer-github/discovery/pkg/worker"
 	"os"
 	"os/signal"
 	"syscall"
@@ -30,33 +28,8 @@ func main() {
 		}
 	}()
 
-	if err := WorkerCommand().ExecuteContext(ctx); err != nil {
+	if err := worker.WorkerCommand().ExecuteContext(ctx); err != nil {
 		fmt.Printf("Error: %v\n", err)
 		os.Exit(1)
 	}
 }
-
-func WorkerCommand() *cobra.Command {
-	cmd := &cobra.Command{
-		RunE: func(cmd *cobra.Command, args []string) error {
-			ctx := cmd.Context()
-			cmd.SilenceUsage = true
-			logger, err := zap.NewProduction()
-			if err != nil {
-				return err
-			}
-
-			w, err := pkg.NewWorker(
-				logger,
-				cmd.Context(),
-			)
-			if err != nil {
-				return err
-			}
-
-			return w.Run(ctx)
-		},
-	}
-
-	return cmd
-}

discovery/pkg/orchestrator/handler.go

@@ -4,13 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/go-errors/errors"
 	model "github.com/opengovern/og-describer-github/discovery/pkg/models"
 	"github.com/opengovern/og-describer-github/discovery/provider"
 	"github.com/opengovern/og-describer-github/global"
 	describe2 "github.com/opengovern/og-util/pkg/describe"
 	"github.com/opengovern/og-util/pkg/es"
-	"github.com/opengovern/og-util/pkg/vault"
 	"go.uber.org/zap"
 	strconv "strconv"
 	"strings"
@@ -46,35 +44,7 @@ func trimJsonFromEmptyObjects(input []byte) ([]byte, error) {
 	return json.Marshal(unknownData)
 }
 
-func Do(ctx context.Context,
-	vlt vault.VaultSourceConfig,
-	logger *zap.Logger,
-	job describe2.DescribeJob,
-	params map[string]string,
-	grpcEndpoint string,
-	describeDeliverToken string,
-	ingestionPipelineEndpoint string,
-	useOpenSearch bool) (resourceIDs []string, err error) {
-	defer func() {
-		if r := recover(); r != nil {
-			err = fmt.Errorf("paniced with error: %v", r)
-			logger.Error("paniced with error", zap.Error(err), zap.String("stackTrace", errors.Wrap(r, 2).ErrorStack()))
-		}
-	}()
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	config, err := vlt.Decrypt(ctx, job.CipherText)
-	if err != nil {
-		return nil, fmt.Errorf("decrypt error: %w", err)
-	}
-	// logger.Info("decrypted config", zap.Any("config", config))
-
-	return doDescribe(ctx, logger, job, params, config, grpcEndpoint, ingestionPipelineEndpoint, describeDeliverToken, useOpenSearch)
-}
-
-func doDescribe(
+func Describe(
 	ctx context.Context,
 	logger *zap.Logger,
 	job describe2.DescribeJob,
@@ -84,7 +54,7 @@ func doDescribe(
 	describeToken string,
 	useOpenSearch bool) ([]string, error) {
 	logger.Info("Making New Resource Sender")
-	rs, err := NewResourceSender(grpcEndpoint, ingestionPipelineEndpoint, describeToken, job.JobID, params,useOpenSearch, logger)
+	rs, err := NewResourceSender(grpcEndpoint, ingestionPipelineEndpoint, describeToken, job.JobID, params, useOpenSearch, logger)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to resource sender: %w", err)
 	}

discovery/pkg/orchestrator/worker.go

@@ -0,0 +1,28 @@
+package task
+
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	maxRetries        = 10
+	retryDelay        = 10 * time.Second
+	connectionRefused = "connect: connection refused"
+)
+
+func retryWithBackoff[T any](operationName string, fetchFunc func() ([]T, error)) ([]T, error) {
+	var result []T
+	var err error
+
+	for attempt := 1; attempt <= maxRetries; attempt++ {
+		result, err = fetchFunc()
+		if err == nil {
+			return result, nil
+		}
+
+		time.Sleep(retryDelay)
+	}
+
+	return nil, fmt.Errorf("failed to complete %s after %d attempts: %w", operationName, maxRetries, err)
+}