feat(dashboard): worker label filters, webhook 429 on limit exhausted

api-contracts/openapi/paths/v1/webhooks/webhook.yaml

@@ -140,6 +140,12 @@ V1WebhookGetDeleteReceiveUpdate:
             schema:
               $ref: "../../../components/schemas/_index.yaml#/APIErrors"
         description: Forbidden
+      "429":
+        content:
+          application/json:
+            schema:
+              $ref: "../../../components/schemas/_index.yaml#/APIErrors"
+        description: Resource limit exceeded for the tenant
     summary: Post a webhook message
     tags:
       - Webhook

api-contracts/openapi/paths/worker/worker.yaml

@@ -35,6 +35,16 @@ withTenant:
           type: array
           items:
             $ref: "../../components/schemas/_index.yaml#/WorkerStatus"
+      - description: Filter by worker labels
+        in: query
+        name: labels
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+            description: The label key-value pairs (delimited by a `:`) to filter by
+            minLength: 1
     responses:
       "200":
         content:

api/v1/server/handlers/v1/webhooks/receive.go

@@ -321,6 +321,16 @@ func (w *V1WebhooksService) V1WebhookReceive(ctx echo.Context, request gen.V1Web
 	)
 
 	if err != nil {
+		if errors.Is(err, repository.ErrResourceExhausted) {
+			return gen.V1WebhookReceive429JSONResponse{
+				Errors: []gen.APIError{
+					{
+						Description: "resource limit exceeded: task run or event limit reached for tenant",
+					},
+				},
+			}, nil
+		}
+
 		return nil, fmt.Errorf("failed to ingest event")
 	}
 

api/v1/server/handlers/workers/list.go

@@ -3,11 +3,13 @@ package workers
 import (
 	"fmt"
 	"math"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
 
+	"github.com/hatchet-dev/hatchet/api/v1/server/oas/apierrors"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers"
 	transformersv1 "github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers/v1"
@@ -30,6 +32,25 @@ func (t *WorkerService) WorkerList(ctx echo.Context, request gen.WorkerListReque
 	}
 }
 
+func parseLabelFilters(labels *[]string) (keys []string, values []string, err error) {
+	if labels == nil {
+		return nil, nil, nil
+	}
+
+	for _, l := range *labels {
+		split := strings.SplitN(l, ":", 2)
+
+		if len(split) != 2 || split[0] == "" || split[1] == "" {
+			return nil, nil, fmt.Errorf("invalid label filter format: %s, expected key:value", l)
+		}
+
+		keys = append(keys, split[0])
+		values = append(values, split[1])
+	}
+
+	return keys, values, nil
+}
+
 func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, request gen.WorkerListRequestObject) (gen.WorkerListResponseObject, error) {
 	reqCtx := ctx.Request().Context()
 	tenantId := tenant.ID
@@ -63,6 +84,13 @@ func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		opts.Statuses = statuses
 	}
 
+	labelKeys, labelValues, err := parseLabelFilters(request.Params.Labels)
+	if err != nil {
+		return gen.WorkerList400JSONResponse(apierrors.NewAPIErrors(err.Error())), nil
+	}
+	opts.LabelKeys = labelKeys
+	opts.LabelValues = labelValues
+
 	_, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v0.list-workers")
 	defer listSpan.End()
 
@@ -152,6 +180,13 @@ func (t *WorkerService) workerListV1(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		opts.Statuses = statuses
 	}
 
+	labelKeys, labelValues, err := parseLabelFilters(request.Params.Labels)
+	if err != nil {
+		return gen.WorkerList400JSONResponse(apierrors.NewAPIErrors(err.Error())), nil
+	}
+	opts.LabelKeys = labelKeys
+	opts.LabelValues = labelValues
+
 	listCtx, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v1.list-workers")
 	defer listSpan.End()
 

frontend/app/src/pages/main/v1/workers/components/worker-columns.tsx

@@ -28,7 +28,7 @@ const startedAtKey: WorkerColumnKeys = 'startedAt';
 const slotsKey: WorkerColumnKeys = 'slots';
 const lastHeartbeatAtKey: WorkerColumnKeys = 'lastHeartbeatAt';
 const runtimeKey: WorkerColumnKeys = 'runtime';
-const labelsKey: WorkerColumnKeys = 'labels';
+export const labelsKey: WorkerColumnKeys = 'labels';
 
 interface WorkerStatusBadgeProps extends BadgeProps {
   status?: string;

frontend/app/src/pages/main/v1/workers/index.tsx

@@ -1,4 +1,9 @@
-import { columns, statusKey, WorkerColumn } from './components/worker-columns';
+import {
+  columns,
+  labelsKey,
+  statusKey,
+  WorkerColumn,
+} from './components/worker-columns';
 import { ToolbarType } from '@/components/v1/molecules/data-table/data-table-toolbar';
 import { DataTable } from '@/components/v1/molecules/data-table/data-table.tsx';
 import { EmptyState } from '@/components/v1/molecules/empty-state/empty-state';
@@ -21,10 +26,12 @@ import { z } from 'zod';
 const workersQuerySchema = z
   .object({
     s: z.array(z.enum(['ACTIVE', 'INACTIVE', 'PAUSED'])).optional(), // status
+    l: z.array(z.string()).optional(), // labels
   })
   .default({})
   .transform((data) => ({
     s: data.s ?? ['ACTIVE', 'PAUSED'],
+    l: data.l ?? [],
   }));
 
 export default function Workers() {
@@ -52,16 +59,19 @@ function WorkersTable() {
   );
 
   const {
-    state: { s: statuses },
+    state: { s: statuses, l: labels },
     columnFilters,
     setColumnFilters,
     resetFilters,
-  } = useZodColumnFilters(workersQuerySchema, paramKey, { s: statusKey });
+  } = useZodColumnFilters(workersQuerySchema, paramKey, {
+    s: statusKey,
+    l: labelsKey,
+  });
 
   const { pagination, setPagination, limit, offset, setPageSize } =
     usePagination({
       key: paramKey,
-      resetPageOnChange: [statuses],
+      resetPageOnChange: [statuses, labels],
     });
 
   const [columnVisibility, setColumnVisibility] =
@@ -82,6 +92,7 @@ function WorkersTable() {
       offset,
       limit,
       statuses: statuses as WorkerStatus[],
+      labels: labels.length > 0 ? labels : undefined,
     }),
     refetchInterval,
   });
@@ -110,6 +121,11 @@ function WorkersTable() {
             { value: 'INACTIVE', label: 'Inactive' },
           ],
         },
+        {
+          columnId: labelsKey,
+          title: 'Labels',
+          type: ToolbarType.KeyValue,
+        },
       ]}
       emptyState={
         <EmptyState

pkg/repository/sqlcv1/workers.sql

@@ -150,6 +150,29 @@ WHERE
             ELSE 'ACTIVE'
         END = ANY(sqlc.narg('statuses')::text[])
     )
+    AND (
+        sqlc.narg('labelKeys')::text[] IS NULL
+        OR sqlc.narg('labelValues')::text[] IS NULL
+        OR (
+            SELECT BOOL_AND(
+                EXISTS (
+                    SELECT 1
+                    FROM "WorkerLabel" wl
+                    WHERE wl."workerId" = workers."id"
+                        AND wl."key" = lf.k
+                        AND (
+                            wl."strValue" = lf.v
+                            OR wl."intValue"::text = lf.v
+                        )
+                )
+            )
+            FROM (
+                SELECT
+                    UNNEST(sqlc.narg('labelKeys')::text[]) AS k,
+                    UNNEST(sqlc.narg('labelValues')::text[]) AS v
+            ) AS lf
+        )
+    )
 ORDER BY
     workers."createdAt" DESC
 OFFSET
@@ -195,6 +218,29 @@ WHERE
             WHEN workers."isPaused" = true THEN 'PAUSED'
             ELSE 'ACTIVE'
         END = ANY(sqlc.narg('statuses')::text[])
+    )
+    AND (
+        sqlc.narg('labelKeys')::text[] IS NULL
+        OR sqlc.narg('labelValues')::text[] IS NULL
+        OR (
+            SELECT BOOL_AND(
+                EXISTS (
+                    SELECT 1
+                    FROM "WorkerLabel" wl
+                    WHERE wl."workerId" = workers."id"
+                        AND wl."key" = lf.k
+                        AND (
+                            wl."strValue" = lf.v
+                            OR wl."intValue"::text = lf.v
+                        )
+                )
+            )
+            FROM (
+                SELECT
+                    UNNEST(sqlc.narg('labelKeys')::text[]) AS k,
+                    UNNEST(sqlc.narg('labelValues')::text[]) AS v
+            ) AS lf
+        )
     );
 
 -- name: GetWorkerById :one