feat: Create & List DLQ job api (#70)

* feat: create DLQ job

* feat: create DLQ job

* Merge

* feat: Create Dlq job api witht testing

* feat: swagger

* feat: add dlqJob name

* fix: mapper

* test: created testing

* feat: list dlq Job

* fix: response createdlqjob

* feat: ListDlqJob api

* feat(dlq): missing project in Create job api

* feat: filter for listDLQ

* test: add error testing for create dlq job

* fix: lint error

* feat(dlq): store dlq information in entropy labels

* fix: lint warning

* feat: should not create dlq job with empty image and prom host

---------

Co-authored-by: Lifosmin Simon <[email protected]>
Co-authored-by: Stewart Jingga <[email protected]>

README.md

@@ -1,3 +1,7 @@
 # dex
 
 Data Experience
+
+## Setting up GCloud Credentials
+1. login to cloud - `gcloud auth login`
+2. setup ADC - `gcloud auth application-default login`

cli/server/server.go

@@ -104,8 +104,7 @@ func runServer(baseCtx context.Context, nrApp *newrelic.Application, zapLog *zap
 	}
 
 	wardenClient := warden.NewClient(cfg.Warden.Addr)
-	dlqConfig := &dlq.DlqJobConfig{
-		// TODO: map cfg.Dlq\
+	dlqConfig := dlq.DlqJobConfig{
 		DlqJobImage:    cfg.Dlq.DlqJobImage,
 		PrometheusHost: cfg.Dlq.PrometheusHost,
 	}

internal/server/server.go

@@ -40,7 +40,7 @@ func Serve(ctx context.Context, addr string,
 	odinAddr string,
 	stencilAddr string,
 	wardenClient *warden.Client,
-	dlqConfig *dlqv1.DlqJobConfig,
+	dlqConfig dlqv1.DlqJobConfig,
 ) error {
 	alertSvc := alertsv1.NewService(sirenClient)
 

internal/server/v1/dlq/errors.go

@@ -5,4 +5,7 @@ import "errors"
 var (
 	ErrFirehoseNamespaceNotFound = errors.New("could not find firehose namespace from resource output")
 	ErrFirehoseNamespaceInvalid  = errors.New("invalid firehose namespace from resource output")
+	ErrFirehoseNotFound          = errors.New("firehose not found")
+	ErrEmptyConfigImage          = errors.New("empty dlq job image")
+	ErrEmptyConfigPrometheusHost = errors.New("empty prometheus host")
 )

internal/server/v1/dlq/fixtures/list_dlq_jobs.json

@@ -0,0 +1,38 @@
+{
+    "dlq_jobs": [
+        {
+            "resource_id": "test-resource-id",
+            "resource_type": "test-resource-type",
+            "topic": "test-topic",
+            "date": "2022-10-21",
+            "name": "test1-firehose-test-topic-2022-10-21",
+            "group": "test-group",
+            "kube_cluster": "test-kube-cluster",
+            "project": "test-project-1",
+            "prometheus_host": "prom_host",
+            "urn": "test-urn-1",
+            "status": "STATUS_UNSPECIFIED",
+            "created_at": "2022-12-10T00:00:00.000Z",
+            "created_by": "[email protected]",
+            "updated_at": "2023-12-10T02:00:00.000Z",
+            "updated_by": "[email protected]"
+        },
+        {
+            "resource_id": "test-resource-id",
+            "resource_type": "test-resource-type",
+            "topic": "test-topic",
+            "date": "2022-10-21",
+            "name": "test2-firehose-test-topic-2022-10-21",
+            "group": "test-group",
+            "kube_cluster": "test-kube-cluster",
+            "project": "test-project-1",
+            "prometheus_host": "prom_host2",
+            "urn": "test-urn-2",
+            "status": "STATUS_UNSPECIFIED",
+            "created_at": "2012-10-10T04:00:00.000Z",
+            "created_by": "[email protected]",
+            "updated_at": "2013-02-12T02:04:00.000Z",
+            "updated_by": "[email protected]"
+        }
+    ]
+}

internal/server/v1/dlq/handler.go

@@ -1,14 +1,17 @@
 package dlq
 
 import (
+	"errors"
 	"log"
 	"net/http"
 
 	entropyv1beta1 "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/entropy/v1beta1"
 	"github.com/go-chi/chi/v5"
 
 	"github.com/goto/dex/entropy"
+	"github.com/goto/dex/generated/models"
 	"github.com/goto/dex/internal/server/gcs"
+	"github.com/goto/dex/internal/server/reqctx"
 	"github.com/goto/dex/internal/server/utils"
 	"github.com/goto/dex/internal/server/v1/firehose"
 )
@@ -55,18 +58,76 @@ func (h *Handler) ListFirehoseDLQ(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
-func (*Handler) listDlqJobs(w http.ResponseWriter, _ *http.Request) {
+func (h *Handler) listDlqJobs(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	labelFilter := map[string]string{}
+	if resourceID := r.URL.Query().Get("resource_id"); resourceID != "" {
+		labelFilter["resource_id"] = resourceID
+	}
+	if resourceType := r.URL.Query().Get("resource_type"); resourceType != "" {
+		labelFilter["resource_type"] = resourceType
+	}
+	if date := r.URL.Query().Get("date"); date != "" {
+		labelFilter["date"] = date
+	}
+
+	dlqJob, err := h.service.ListDlqJob(ctx, labelFilter)
+	if err != nil {
+		if errors.Is(err, ErrFirehoseNotFound) {
+			utils.WriteErrMsg(w, http.StatusNotFound, err.Error())
+			return
+		}
+		utils.WriteErr(w, err)
+		return
+	}
+
 	utils.WriteJSON(w, http.StatusOK, map[string]interface{}{
-		"dlq_jobs": []interface{}{},
-	})
+		"dlq_jobs": dlqJob,
+	},
+	)
 }
 
-func (*Handler) createDlqJob(w http.ResponseWriter, _ *http.Request) {
-	// transform request body into DlqJob (validation?)
-	// call service.CreateDLQJob
+func (h *Handler) createDlqJob(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	reqCtx := reqctx.From(ctx)
+	if reqCtx.UserEmail == "" {
+		utils.WriteErrMsg(w, http.StatusUnauthorized, "user header is required")
+		return
+	}
+
+	var body models.DlqJobForm
+	if err := utils.ReadJSON(r, &body); err != nil {
+		utils.WriteErr(w, err)
+		return
+	}
+	if err := body.Validate(nil); err != nil {
+		utils.WriteErrMsg(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	dlqJob := models.DlqJob{
+		BatchSize:    *body.BatchSize,
+		Date:         *body.Date,
+		ErrorTypes:   body.ErrorTypes,
+		NumThreads:   *body.NumThreads,
+		ResourceID:   *body.ResourceID,
+		ResourceType: *body.ResourceType,
+		Topic:        *body.Topic,
+	}
+
+	updatedDlqJob, err := h.service.CreateDLQJob(ctx, reqCtx.UserEmail, dlqJob)
+	if err != nil {
+		if errors.Is(err, ErrFirehoseNotFound) {
+			utils.WriteErrMsg(w, http.StatusNotFound, err.Error())
+			return
+		}
+		utils.WriteErr(w, err)
+		return
+	}
 
 	utils.WriteJSON(w, http.StatusOK, map[string]interface{}{
-		"dlq_job": nil,
+		"dlq_job": updatedDlqJob,
 	})
 }