Merge branch 'DFEXP-118' of github.com:goto/dex into DFEXP-118

internal/server/v1/dlq/errors.go

@@ -6,5 +6,4 @@ 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")
-	ErrInternal                  = errors.New("internal_error")
 )

internal/server/v1/dlq/handler.go

@@ -89,17 +89,33 @@ func (h *Handler) listDlqJobs(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *Handler) createDlqJob(w http.ResponseWriter, r *http.Request) {
-	// transform request body into DlqJob (validation?)
 	ctx := r.Context()
 	reqCtx := reqctx.From(ctx)
-	var dlqJob models.DlqJob
+	if reqCtx.UserEmail == "" {
+		utils.WriteErrMsg(w, http.StatusUnauthorized, "user header is required")
+		return
+	}
 
-	if err := utils.ReadJSON(r, &dlqJob); err != nil {
+	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,
+	}
 
-	// call service.CreateDLQJob
 	err := h.service.CreateDLQJob(ctx, reqCtx.UserEmail, &dlqJob)
 	if err != nil {
 		if errors.Is(err, ErrFirehoseNotFound) {
@@ -109,7 +125,7 @@ func (h *Handler) createDlqJob(w http.ResponseWriter, r *http.Request) {
 		utils.WriteErr(w, err)
 		return
 	}
-	// return
+
 	utils.WriteJSON(w, http.StatusOK, map[string]interface{}{
 		"dlq_job": dlqJob,
 	})

internal/server/v1/dlq/mapper.go

@@ -49,7 +49,7 @@ func enrichDlqJob(job *models.DlqJob, res *entropyv1beta1.Resource, cfg DlqJobCo
 	job.Status = status
 	job.CreatedAt = strfmt.DateTime(res.CreatedAt.AsTime())
 	job.UpdatedAt = strfmt.DateTime(res.UpdatedAt.AsTime())
-
+	job.Project = res.Project
 	job.KubeCluster = kubeCluster
 	job.ContainerImage = cfg.DlqJobImage
 	job.PrometheusHost = cfg.PrometheusHost

internal/server/v1/dlq/service.go

@@ -59,12 +59,12 @@ func (s *Service) CreateDLQJob(ctx context.Context, userEmail string, dlqJob *mo
 	entropyCtx := metadata.AppendToOutgoingContext(ctx, "user-id", userEmail)
 	rpcReq := &entropyv1beta1.CreateResourceRequest{Resource: res}
 	rpcResp, err := s.client.CreateResource(entropyCtx, rpcReq)
-	dlqJob.Urn = rpcResp.Resource.Urn
 	if err != nil {
-		outErr := ErrInternal
-		return outErr
+		return err
 	}
 
+	dlqJob.Urn = rpcResp.Resource.Urn
+
 	return nil
 }
 

swagger.yml

@@ -1347,6 +1347,37 @@ definitions:
         type: string
       dlq_gcs_credential_path:
         type: string
+  DlqJobForm:
+    type: object
+    required:
+      - date
+      - topic
+      - resource_id
+      - resource_type
+      - batch_size
+      - num_threads
+    properties:
+      date:
+        type: string
+        example: "2012-10-30"
+        minLength: 1
+      topic:
+        type: string
+        minLength: 1
+      resource_id:
+        type: string
+        minLength: 1
+      resource_type:
+        type: string
+        enum:
+          - firehose
+      error_types:
+        type: string
+        description: "List of firehose error types, comma separated"
+      batch_size:
+        type: integer
+      num_threads:
+        type: integer
   Subscription:
     type: object
     description: "Siren subscription model: https://github.com/goto/siren/blob/v0.6.4/proto/siren.swagger.yaml#L1325"