Merge pull request #1194 from BudEcosystem/feat/pipeline-deployment-actions

feat(budpipeline): add deployment actions with internal auth support

Author: dittops

infra/helm/bud/templates/microservices/budpipeline.yaml

@@ -40,7 +40,7 @@ spec:
       - name: budpipeline
         image: {{ .Values.microservices.budpipeline.image }}
         imagePullPolicy: {{ .Values.imagePullPolicy }}
-        command: ["sh", "-c", "uvicorn budpipeline.main:app --host 0.0.0.0 --port 8010"]
+        command: ["sh", "-c", "alembic upgrade head && uvicorn budpipeline.main:app --host 0.0.0.0 --port 8010"]
         ports:
         - containerPort: 8010
         envFrom:

services/budadmin/src/components/pipelineEditor/components/ActionConfigPanel.tsx

@@ -243,6 +243,7 @@ export interface ActionConfigPanelProps {
     projects?: SelectOption[];
     endpoints?: SelectOption[];
     providers?: SelectOption[];
+    credentials?: SelectOption[];
   };
   /** Whether data sources are loading */
   loadingDataSources?: Set<string>;
@@ -547,6 +548,8 @@ export function ActionConfigPanel({
           return dataSources.endpoints || [];
         case 'provider_ref':
           return dataSources.providers || [];
+        case 'credential_ref':
+          return dataSources.credentials || [];
         default:
           return [];
       }
@@ -664,8 +667,8 @@ export function ActionConfigPanel({
             </select>
           )}
 
-          {/* Ref types (model, cluster, provider, etc.) - with toggle for template mode */}
-          {['model_ref', 'cluster_ref', 'project_ref', 'endpoint_ref', 'provider_ref'].includes(param.type) && (
+          {/* Ref types (model, cluster, provider, credential, etc.) - with toggle for template mode */}
+          {['model_ref', 'cluster_ref', 'project_ref', 'endpoint_ref', 'provider_ref', 'credential_ref'].includes(param.type) && (
             <div>
               <div style={{ display: 'flex', gap: '8px', marginBottom: '6px' }}>
                 <button

services/budadmin/src/flows/components/CommonStatus.tsx

@@ -156,9 +156,7 @@ export default function CommonStatus({
     }, [steps, workflowId]);
 
     useEffect(() => {
-        console.log(`socket`, socket)
         if (socket) {
-            console.log(handleNotification)
             socket.on("notification_received", handleNotification);
         }
 

services/budadmin/src/pages/home/budpipelines/[id]/index.tsx

@@ -26,6 +26,8 @@ import { useCluster } from "src/hooks/useCluster";
 import { useModels } from "src/hooks/useModels";
 import { useCloudProviders } from "src/hooks/useCloudProviders";
 import { useProjects } from "src/hooks/useProjects";
+import { useProprietaryCredentials } from "src/stores/useProprietaryCredentials";
+import { useEndPoints } from "src/hooks/useEndPoint";
 import { PrimaryButton } from "@/components/ui/bud/form/Buttons";
 import { useDrawer } from "src/hooks/useDrawer";
 import StepDetailDrawer from "@/components/pipelineEditor/components/StepDetailDrawer";
@@ -121,6 +123,8 @@ const WorkflowDetail = () => {
   const { models, getGlobalModels } = useModels();
   const { providers, getProviders } = useCloudProviders();
   const { projects, getProjects } = useProjects();
+  const { credentials: proprietaryCredentials, getCredentials: getProprietaryCredentials } = useProprietaryCredentials();
+  const { endPoints, getEndPoints } = useEndPoints();
 
   const [activeTab, setActiveTab] = useState("dag");
   const [selectedStep, setSelectedStep] = useState<PipelineStep | null>(null);
@@ -144,11 +148,13 @@ const WorkflowDetail = () => {
     [executions]
   );
 
-  // Transform clusters, models, projects, and providers to SelectOption format for WorkflowEditor
+  // Transform clusters, models, projects, providers, credentials, and endpoints to SelectOption format for WorkflowEditor
+  // Note: clusters use cluster_id (budcluster UUID) as value, not id (budapp UUID),
+  // to match what deploy-workflow endpoint expects
   const dataSources = useMemo(() => ({
     clusters: clusters.map((c) => ({
       label: c.name,
-      value: c.id,
+      value: c.cluster_id,
     })),
     models: models.map((m) => ({
       label: m.name,
@@ -162,9 +168,17 @@ const WorkflowDetail = () => {
       label: p.name,
       value: p.id,
     })),
-  }), [clusters, models, projects, providers]);
+    credentials: proprietaryCredentials.map((c) => ({
+      label: `${c.name} (${c.type})`,
+      value: c.id,
+    })),
+    endpoints: endPoints.map((e) => ({
+      label: e.name || e.id || "",
+      value: e.id || "",
+    })),
+  }), [clusters, models, projects, providers, proprietaryCredentials, endPoints]);
 
-  // Fetch clusters, models, projects, and providers when entering edit mode
+  // Fetch clusters, models, projects, providers, credentials, and endpoints when entering edit mode
   useEffect(() => {
     if (isEditing) {
       const loadingSet = new Set<string>();
@@ -181,6 +195,12 @@ const WorkflowDetail = () => {
       if (providers.length === 0) {
         loadingSet.add("providers");
       }
+      if (proprietaryCredentials.length === 0) {
+        loadingSet.add("credentials");
+      }
+      if (endPoints.length === 0) {
+        loadingSet.add("endpoints");
+      }
 
       if (loadingSet.size > 0) {
         setLoadingDataSources(loadingSet);
@@ -242,6 +262,33 @@ const WorkflowDetail = () => {
           );
         }
 
+        if (proprietaryCredentials.length === 0) {
+          fetchPromises.push(
+            (async () => {
+              await getProprietaryCredentials({ page: 1, limit: 100 });
+              setLoadingDataSources(prev => {
+                const next = new Set(prev);
+                next.delete("credentials");
+                return next;
+              });
+            })()
+          );
+        }
+
+        if (endPoints.length === 0) {
+          fetchPromises.push(
+            (async () => {
+              // Fetch all endpoints (no project filter needed for pipeline actions)
+              await getEndPoints({ page: 1, limit: 100 });
+              setLoadingDataSources(prev => {
+                const next = new Set(prev);
+                next.delete("endpoints");
+                return next;
+              });
+            })()
+          );
+        }
+
         Promise.all(fetchPromises).finally(() => {
           setLoadingDataSources(new Set());
         });

services/budadmin/src/pages/home/deployments/[slug]/settings/AutoscaleSettings.tsx

@@ -479,7 +479,7 @@ export const AutoscaleSettings: React.FC<AutoscaleSettingsProps> = ({
                 </label>
                 <InputNumber
                   value={config.minReplicas}
-                  onChange={(value) => updateConfig({ minReplicas: value || 1 })}
+                  onChange={(value) => updateConfig({ minReplicas: value ?? 1 })}
                   min={0}
                   max={config.maxReplicas}
                   style={{ ...inputStyle, width: "100%" }}
@@ -501,7 +501,7 @@ export const AutoscaleSettings: React.FC<AutoscaleSettingsProps> = ({
                 </label>
                 <InputNumber
                   value={config.maxReplicas}
-                  onChange={(value) => updateConfig({ maxReplicas: value || 10 })}
+                  onChange={(value) => updateConfig({ maxReplicas: value ?? 10 })}
                   min={config.minReplicas}
                   max={100}
                   style={{ ...inputStyle, width: "100%" }}

services/budadmin/src/types/actions.ts

@@ -22,7 +22,8 @@ export type ParamType =
   | 'cluster_ref'
   | 'project_ref'
   | 'endpoint_ref'
-  | 'provider_ref';
+  | 'provider_ref'
+  | 'credential_ref';
 
 export interface SelectOption {
   value: string;

services/budapp/budapp/cluster_ops/schemas.py

@@ -605,6 +605,10 @@ class RecommendedClusterRequest(BaseModel):
         default=None,
         description="Comma-separated model endpoint types (e.g., 'EMBEDDING', 'LLM') for engine selection",
     )
+    debug: bool = Field(
+        default=False,
+        description="If True, run simulation synchronously and wait for completion before returning",
+    )
 
 
 class GrafanaDashboardResponse(SuccessResponse):

services/budapp/budapp/endpoint_ops/endpoint_routes.py

@@ -19,21 +19,22 @@
 from typing import List, Optional, Union
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, Header, Query, status
+from fastapi import APIRouter, Body, Depends, Header, Query, status
 from sqlalchemy.orm import Session
 from typing_extensions import Annotated
 
 from budapp.commons import logging
 from budapp.commons.dependencies import (
     get_current_active_user,
+    get_current_active_user_or_internal,
     get_session,
     parse_ordering_fields,
 )
 from budapp.commons.exceptions import ClientException
 from budapp.user_ops.schemas import User
 
 from ..commons.constants import PermissionEnum
-from ..commons.permission_handler import require_permissions
+from ..commons.permission_handler import require_permissions, require_permissions_or_internal
 from ..commons.schemas import ErrorResponse, SuccessResponse
 from ..workflow_ops.schemas import RetrieveWorkflowDataResponse
 from ..workflow_ops.services import WorkflowService
@@ -46,6 +47,7 @@
     DeleteWorkerRequest,
     DeploymentPricingResponse,
     DeploymentSettingsResponse,
+    EndpointDeleteResponse,
     EndpointFilter,
     EndpointPaginatedResponse,
     ModelClusterDetailResponse,
@@ -146,28 +148,37 @@ async def list_all_endpoints(
             "description": "Invalid request parameters",
         },
         status.HTTP_200_OK: {
-            "model": SuccessResponse,
+            "model": EndpointDeleteResponse,
             "description": "Successfully executed delete endpoint workflow",
         },
     },
     description="Delete an endpoint by ID",
 )
-@require_permissions(permissions=[PermissionEnum.ENDPOINT_MANAGE])
+@require_permissions_or_internal(permissions=[PermissionEnum.ENDPOINT_MANAGE])
 async def delete_endpoint(
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     endpoint_id: UUID,
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,
-) -> Union[SuccessResponse, ErrorResponse]:
-    """Delete a endpoint by its ID."""
+    callback_topic: Annotated[Optional[str], Body(embed=True)] = None,
+) -> Union[EndpointDeleteResponse, ErrorResponse]:
+    """Delete a endpoint by its ID.
+
+    Args:
+        callback_topic: Optional Dapr pub/sub topic for budpipeline integration.
+                       If provided, completion events will be published to this topic.
+    """
     try:
-        db_workflow = await EndpointService(session).delete_endpoint(endpoint_id, current_user.id)
+        db_workflow = await EndpointService(session).delete_endpoint(
+            endpoint_id, current_user.id, callback_topic=callback_topic
+        )
         logger.debug(f"Endpoint deleting initiated with workflow id: {db_workflow.id}")
-        return SuccessResponse(
+        return EndpointDeleteResponse(
             message="Deployment deleting initiated successfully",
             code=status.HTTP_200_OK,
             object="endpoint.delete",
+            workflow_id=db_workflow.id,
         )
     except ClientException as e:
         logger.exception(f"Failed to delete endpoint: {e}")
@@ -1018,10 +1029,10 @@ async def update_deployment_settings(
     },
     description="Get autoscale configuration for an endpoint",
 )
-@require_permissions(permissions=[PermissionEnum.ENDPOINT_VIEW])
+@require_permissions_or_internal(permissions=[PermissionEnum.ENDPOINT_VIEW])
 async def get_autoscale_config(
     endpoint_id: UUID,
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,
@@ -1065,11 +1076,11 @@ async def get_autoscale_config(
     },
     description="Update autoscale configuration for an endpoint",
 )
-@require_permissions(permissions=[PermissionEnum.ENDPOINT_MANAGE])
+@require_permissions_or_internal(permissions=[PermissionEnum.ENDPOINT_MANAGE])
 async def update_autoscale_config(
     endpoint_id: UUID,
     request: UpdateAutoscaleRequest,
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,

services/budapp/budapp/endpoint_ops/schemas.py

@@ -100,6 +100,15 @@ class EndpointPaginatedResponse(PaginatedSuccessResponse):
     endpoints: list[EndpointListResponse] = []
 
 
+class EndpointDeleteResponse(SuccessResponse):
+    """Response schema for endpoint delete operation.
+
+    Includes the workflow_id for tracking the async delete workflow.
+    """
+
+    workflow_id: Optional[UUID] = None
+
+
 class WorkerInfoFilter(BaseModel):
     """Filter for worker info."""