fix: recover unhandled exceptions

Author: Arta Asadi

discovery/pkg/nats.go

@@ -1,22 +1,11 @@
 package pkg
 
 import (
-	"context"
-	"encoding/json"
-	"errors"
-	"github.com/opengovern/og-describer-kubernetes/discovery/pkg/orchestrator"
-	"github.com/opengovern/og-describer-kubernetes/global"
-	"os"
-	"runtime"
-	"time"
-
-	"github.com/nats-io/nats.go/jetstream"
-
-	"github.com/opengovern/og-util/pkg/describe"
 	esSinkClient "github.com/opengovern/og-util/pkg/es/ingest/client"
 	"github.com/opengovern/og-util/pkg/jq"
 	"github.com/opengovern/og-util/pkg/opengovernance-es-sdk"
 	"go.uber.org/zap"
+	"os"
 )
 
 type Worker struct {
@@ -30,103 +19,3 @@ type Worker struct {
 var (
 	ManualTriggers = os.Getenv("MANUAL_TRIGGERS")
 )
-
-func NewWorker(
-
-	logger *zap.Logger,
-	ctx context.Context,
-) (*Worker, error) {
-	url := os.Getenv("NATS_URL")
-	jq, err := jq.New(url, logger)
-	if err != nil {
-		logger.Error("failed to create job queue", zap.Error(err), zap.String("url", url))
-		return nil, err
-	}
-
-	topic := global.JobQueueTopic
-	if ManualTriggers == "true" {
-		topic = global.JobQueueTopicManuals
-	}
-	if err := jq.Stream(ctx, global.StreamName, " describe job runner queue", []string{topic}, 5000); err != nil {
-		logger.Error("failed to create stream", zap.Error(err))
-		return nil, err
-	}
-
-	w := &Worker{
-		logger: logger,
-		jq:     jq,
-	}
-
-	return w, nil
-}
-
-func (w *Worker) Run(ctx context.Context) error {
-	w.logger.Info("starting to consume")
-	topic := global.JobQueueTopic
-	consumer := global.ConsumerGroup
-	if ManualTriggers == "true" {
-		topic = global.JobQueueTopicManuals
-		consumer = global.ConsumerGroupManuals
-	}
-	consumeCtx, err := w.jq.ConsumeWithConfig(ctx, consumer, global.StreamName, []string{topic}, jetstream.ConsumerConfig{
-		Replicas:          1,
-		AckPolicy:         jetstream.AckExplicitPolicy,
-		DeliverPolicy:     jetstream.DeliverAllPolicy,
-		MaxAckPending:     -1,
-		AckWait:           time.Minute * 30,
-		InactiveThreshold: time.Hour,
-	}, []jetstream.PullConsumeOpt{
-		jetstream.PullMaxMessages(1),
-	}, func(msg jetstream.Msg) {
-		w.logger.Info("received a new job")
-
-		defer msg.Ack()
-
-		ctx, cancel := context.WithTimeoutCause(ctx, time.Minute*25, errors.New("describe worker timed out"))
-		defer cancel()
-
-		if err := w.ProcessMessage(ctx, msg); err != nil {
-			w.logger.Error("failed to process message", zap.Error(err))
-		}
-		err := msg.Ack()
-		if err != nil {
-			w.logger.Error("failed to ack message", zap.Error(err))
-		}
-
-		w.logger.Info("processing a job completed")
-	})
-	if err != nil {
-		return err
-	}
-
-	w.logger.Info("consuming")
-
-	<-ctx.Done()
-	consumeCtx.Drain()
-	consumeCtx.Stop()
-
-	return nil
-}
-
-func (w *Worker) ProcessMessage(ctx context.Context, msg jetstream.Msg) error {
-	startTime := time.Now()
-	var input describe.DescribeWorkerInput
-	err := json.Unmarshal(msg.Data(), &input)
-	if err != nil {
-		return err
-	}
-	runtime.GC()
-
-	w.logger.Info("running job", zap.Uint("id", input.DescribeJob.JobID), zap.String("type", input.DescribeJob.ResourceType), zap.String("providerID", input.DescribeJob.ProviderID))
-
-	err = orchestrator.DescribeHandler(ctx, w.logger, orchestrator.TriggeredByLocal, input)
-	endTime := time.Now()
-
-	w.logger.Info("job completed", zap.Uint("id", input.DescribeJob.JobID), zap.String("type", input.DescribeJob.ResourceType), zap.String("providerID", input.DescribeJob.ProviderID), zap.Duration("duration", endTime.Sub(startTime)))
-	if err != nil {
-		w.logger.Error("failure while running job", zap.Error(err))
-		return err
-	}
-
-	return nil
-}

discovery/pkg/orchestrator/handler.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/go-errors/errors"
 	model "github.com/opengovern/og-describer-kubernetes/discovery/pkg/models"
 	"github.com/opengovern/og-describer-kubernetes/discovery/provider"
 	"github.com/opengovern/og-describer-kubernetes/global"
 	"github.com/opengovern/og-describer-kubernetes/global/constants"
 	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"
@@ -47,34 +45,6 @@ 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 Describe(ctx, logger, job, params, config, grpcEndpoint, ingestionPipelineEndpoint, describeDeliverToken, useOpenSearch)
-}
-
 func Describe(
 	ctx context.Context,
 	logger *zap.Logger,