[Platform API] Get and wait for Jobs & Reload App Service Configuration (#105)

Author: quaark

pkg/controlplane/http/session.go

@@ -30,6 +30,7 @@ import (
 	"github.com/v3io/v3io-go/pkg/controlplane"
 	"github.com/v3io/v3io-go/pkg/errors"
 
+	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
 	"github.com/valyala/fasthttp"
 )
@@ -245,6 +246,115 @@ func (s *session) GetRunningUserAttributesSync(getRunningUserAttributesInput *v3
 	return &userNameOutput, err
 }
 
+// ReloadAppServicesConfigAndWaitForCompletion reloads the app service config in the backend and waits for job completion (blocking)
+func (s *session) ReloadAppServicesConfigAndWaitForCompletion(ctx context.Context, retryInterval, timeout time.Duration) error {
+	jobId, err := s.ReloadAppServicesConfig(ctx)
+	if err != nil {
+		return errors.Wrap(err, "Failed reloading app service config")
+	}
+
+	return s.WaitForJobCompletion(ctx, jobId, retryInterval, timeout)
+}
+
+// ReloadAppServicesConfig reloads the app service config in the backend (blocking)
+func (s *session) ReloadAppServicesConfig(ctx context.Context) (string, error) {
+	reloadAppServicesConfigInput := v3ioc.ReloadAppServicesConfigInput{
+		ControlPlaneInput: v3ioc.ControlPlaneInput{
+			Ctx: ctx,
+		},
+	}
+
+	reloadAppServicesConfigJobOutput := v3ioc.ReloadAppServicesConfigJobOutput{}
+
+	err := s.createResource(ctx,
+		"configurations/app_services/reloads",
+		"cluster_configuration_reload",
+		&reloadAppServicesConfigInput.ControlPlaneInput,
+		map[string]string{},
+		&reloadAppServicesConfigJobOutput.ControlPlaneOutput,
+		&reloadAppServicesConfigJobOutput.JobAttributes)
+
+	if err != nil {
+		return "", err
+	}
+
+	return reloadAppServicesConfigJobOutput.ID, nil
+}
+
+// WaitForJobCompletion waits for completion of job with given id (blocking)
+func (s *session) WaitForJobCompletion(ctx context.Context, jobId string, retryInterval, timeout time.Duration) error {
+	getJobsInput := v3ioc.GetJobsInput{
+		ControlPlaneInput: v3ioc.ControlPlaneInput{
+			ID:  jobId,
+			Ctx: ctx,
+		},
+	}
+
+	deadline := time.Now().Add(timeout)
+
+	for time.Now().Before(deadline) {
+		getJobsOutput, err := s.GetJobs(&getJobsInput)
+		if err != nil {
+			return errors.Wrap(err, "Failed getting job")
+		}
+
+		switch getJobsOutput.State {
+		case v3ioc.JobStateCompleted:
+			s.logger.DebugWithCtx(ctx,
+				"Job Completed",
+				"jobId", jobId,
+				"jobResult", getJobsOutput.Result)
+			return nil
+		case v3ioc.JobStateFailed:
+			s.logger.WarnWithCtx(ctx,
+				"Job has failed",
+				"jobId", jobId,
+				"jobResult", getJobsOutput.Result)
+			return errors.New(
+				"Job has failed")
+		case v3ioc.JobStateCanceled:
+			s.logger.WarnWithCtx(ctx,
+				"Job was canceled",
+				"jobId", jobId,
+				"jobResult", getJobsOutput.Result)
+			return errors.New("Job was canceled")
+		default:
+			s.logger.DebugWithCtx(ctx,
+				"Job in progress",
+				"jobId", jobId,
+				"retryInterval", retryInterval,
+				"timeout", timeout)
+
+			// TODO: create and use backoff
+			time.Sleep(retryInterval)
+		}
+	}
+
+	return errors.New("Timed out waiting for job completion")
+}
+
+// GetJobs gets jobs (blocking)
+func (s *session) GetJobs(getJobsInput *v3ioc.GetJobsInput) (*v3ioc.GetJobsOutput, error) {
+
+	// prepare job response resource
+	getJobsOutput := v3ioc.GetJobsOutput{}
+
+	// specific path for job detail endpoint
+	detailPath := fmt.Sprintf("jobs/%s", getJobsInput.ID)
+
+	err := s.getResource(getJobsInput.Ctx,
+		detailPath,
+		&getJobsInput.ControlPlaneInput,
+		&getJobsOutput.ControlPlaneOutput,
+		&getJobsOutput.JobAttributes)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &getJobsOutput, err
+}
+
 func (s *session) createResource(ctx context.Context,
 	path string,
 	kind string,

pkg/controlplane/resources.go

@@ -166,3 +166,40 @@ const (
 	SessionKind   Kind = "session"
 	AccessKeyKind Kind = "accessKey"
 )
+
+type JobState string
+
+const (
+	JobStateCreated      JobState = "created"
+	JobStateDispatched   JobState = "dispatched"
+	JobStateInProgress   JobState = "in_progress"
+	JobStateCompleted    JobState = "completed"
+	JobStateCanceled     JobState = "canceled"
+	JobStateRepublishing JobState = "republishing"
+	JobStateFailed       JobState = "failed"
+)
+
+type JobWebHook struct {
+	Method   string `json:"method,omitempty"`
+	Target   string `json:"target,omitempty"`
+	Resource string `json:"resource,omitempty"`
+	Status   int    `json:"status,omitempty"`
+	Payload  string `json:"payload,omitempty"`
+}
+
+type JobAttributes struct {
+	Kind                   string       `json:"kind,omitempty"`
+	Params                 string       `json:"params,omitempty"`
+	UIFields               string       `json:"ui_fields,omitempty"`
+	MaxTotalExecutionTime  int          `json:"max_total_execution_time,omitempty"`
+	MaxWorkerExecutionTime int          `json:"max_worker_execution_time,omitempty"`
+	Delay                  float64      `json:"delay,omitempty"`
+	State                  JobState     `json:"state,omitempty"`
+	Result                 string       `json:"result,omitempty"`
+	CreatedAt              string       `json:"created_at,omitempty"`
+	OnSuccess              []JobWebHook `json:"on_success,omitempty"`
+	OnFailure              []JobWebHook `json:"on_failure,omitempty"`
+	LinkedResources        string       `json:"linked_resources,omitempty"`
+	UpdatedAt              string       `json:"updated_at,omitempty"`
+	Handler                string       `json:"handler,omitempty"`
+}

pkg/controlplane/types.go

@@ -49,6 +49,18 @@ type Session interface {
 
 	// GetRunningUserAttributesSync returns user's attributes related to session's access key (blocking)
 	GetRunningUserAttributesSync(*GetRunningUserAttributesInput) (*GetRunningUserAttributesOutput, error)
+
+	// ReloadAppServicesConfigAndWaitForCompletion reloads the app service config in the backend and waits for job completion (blocking)
+	ReloadAppServicesConfigAndWaitForCompletion(ctx context.Context, retryInterval, timeout time.Duration) error
+
+	// ReloadAppServicesConfig reloads the app service config in the backend (blocking)
+	ReloadAppServicesConfig(ctx context.Context) (string, error)
+
+	// WaitForJobCompletion waits for completion of job with given id (blocking)
+	WaitForJobCompletion(ctx context.Context, jobId string, retryInterval, timeout time.Duration) error
+
+	// GetJobs gets jobs (blocking)
+	GetJobs(getJobsInput *GetJobsInput) (*GetJobsOutput, error)
 }
 
 type ControlPlaneInput struct {
@@ -151,3 +163,25 @@ type GetRunningUserAttributesOutput struct {
 	ControlPlaneOutput
 	UserAttributes
 }
+
+// GetJobsInput specifies how to get a job
+type GetJobsInput struct {
+	ControlPlaneInput
+}
+
+// GetJobsOutput specifies holds the response from get jobs
+type GetJobsOutput struct {
+	ControlPlaneOutput
+	JobAttributes
+}
+
+// ReloadAppServicesConfigInput specifies how to reload the app services configuration
+type ReloadAppServicesConfigInput struct {
+	ControlPlaneInput
+}
+
+// ReloadAppServicesConfigJobOutput specifies holds the response from reload app services config
+type ReloadAppServicesConfigJobOutput struct {
+	ControlPlaneOutput
+	JobAttributes
+}