workflow.go

package adminservice

import (
	"context"
	"time"

	"github.com/lyft/flyteadmin/pkg/audit"

	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/core"
	"github.com/lyft/flytestdlib/logger"

	"github.com/lyft/flyteadmin/pkg/rpc/adminservice/util"
	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/admin"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func (m *AdminService) CreateWorkflow(
	ctx context.Context,
	request *admin.WorkflowCreateRequest) (*admin.WorkflowCreateResponse, error) {
	defer m.interceptPanic(ctx, request)
	requestedAt := time.Now()
	if request == nil {
		return nil, status.Errorf(codes.InvalidArgument, "Incorrect request, nil requests not allowed")
	}
	var response *admin.WorkflowCreateResponse
	var err error
	m.Metrics.workflowEndpointMetrics.create.Time(func() {
		response, err = m.WorkflowManager.CreateWorkflow(ctx, *request)
	})
	audit.NewLogBuilder().WithAuthenticatedCtx(ctx).WithRequest(
		"CreateWorkflow",
		audit.ParametersFromIdentifier(request.Id),
		audit.ReadWrite,
		requestedAt,
	).WithResponse(time.Now(), err).Log(ctx)
	if err != nil {
		return nil, util.TransformAndRecordError(err, &m.Metrics.workflowEndpointMetrics.create)
	}
	m.Metrics.workflowEndpointMetrics.create.Success()
	return response, nil
}

func (m *AdminService) GetWorkflow(ctx context.Context, request *admin.ObjectGetRequest) (*admin.Workflow, error) {
	defer m.interceptPanic(ctx, request)
	requestedAt := time.Now()
	if request == nil {
		return nil, status.Errorf(codes.InvalidArgument, "Incorrect request, nil requests not allowed")
	}
	// NOTE: When the Get HTTP endpoint is called the resource type is implicit (from the URL) so we must add it
	// to the request.
	if request.Id != nil && request.Id.ResourceType == core.ResourceType_UNSPECIFIED {
		logger.Info(ctx, "Adding resource type for unspecified value in request: [%+v]", request)
		request.Id.ResourceType = core.ResourceType_WORKFLOW
	}
	var response *admin.Workflow
	var err error
	m.Metrics.workflowEndpointMetrics.get.Time(func() {
		response, err = m.WorkflowManager.GetWorkflow(ctx, *request)
	})
	audit.NewLogBuilder().WithAuthenticatedCtx(ctx).WithRequest(
		"GetWorkflow",
		audit.ParametersFromIdentifier(request.Id),
		audit.ReadOnly,
		requestedAt,
	).WithResponse(time.Now(), err).Log(ctx)
	if err != nil {
		return nil, util.TransformAndRecordError(err, &m.Metrics.workflowEndpointMetrics.get)
	}
	m.Metrics.workflowEndpointMetrics.get.Success()
	return response, nil
}

func (m *AdminService) ListWorkflowIds(ctx context.Context, request *admin.NamedEntityIdentifierListRequest) (
	*admin.NamedEntityIdentifierList, error) {
	defer m.interceptPanic(ctx, request)
	requestedAt := time.Now()
	if request == nil {
		return nil, status.Errorf(codes.InvalidArgument, "Incorrect request, nil requests not allowed")
	}

	var response *admin.NamedEntityIdentifierList
	var err error
	m.Metrics.workflowEndpointMetrics.listIds.Time(func() {
		response, err = m.WorkflowManager.ListWorkflowIdentifiers(ctx, *request)
	})
	audit.NewLogBuilder().WithAuthenticatedCtx(ctx).WithRequest(
		"ListWorkflowIds",
		map[string]string{
			audit.Project: request.Project,
			audit.Domain:  request.Domain,
		},
		audit.ReadOnly,
		requestedAt,
	).WithResponse(time.Now(), err).Log(ctx)
	if err != nil {
		return nil, util.TransformAndRecordError(err, &m.Metrics.workflowEndpointMetrics.listIds)
	}

	m.Metrics.workflowEndpointMetrics.listIds.Success()
	return response, nil
}

func (m *AdminService) ListWorkflows(ctx context.Context, request *admin.ResourceListRequest) (*admin.WorkflowList, error) {
	defer m.interceptPanic(ctx, request)
	requestedAt := time.Now()
	if request == nil {
		return nil, status.Errorf(codes.InvalidArgument, "Incorrect request, nil requests not allowed")
	}
	var response *admin.WorkflowList
	var err error
	m.Metrics.workflowEndpointMetrics.list.Time(func() {
		response, err = m.WorkflowManager.ListWorkflows(ctx, *request)
	})
	audit.NewLogBuilder().WithAuthenticatedCtx(ctx).WithRequest(
		"ListWorkflows",
		audit.ParametersFromNamedEntityIdentifier(request.Id),
		audit.ReadOnly,
		requestedAt,
	).WithResponse(time.Now(), err).Log(ctx)
	if err != nil {
		return nil, util.TransformAndRecordError(err, &m.Metrics.workflowEndpointMetrics.list)
	}
	m.Metrics.workflowEndpointMetrics.list.Success()
	return response, nil
}