oc-lib/models/workflow/workflow.go

package workflow

import (
	"errors"

	"cloud.o-forge.io/core/oc-lib/models/peer"
	"cloud.o-forge.io/core/oc-lib/models/resources"
	"cloud.o-forge.io/core/oc-lib/models/resources/compute"
	"cloud.o-forge.io/core/oc-lib/models/resources/storage"
	"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
	"cloud.o-forge.io/core/oc-lib/models/utils"
	"cloud.o-forge.io/core/oc-lib/tools"
)

/*
* AbstractWorkflow is a struct that represents a workflow for resource or native workflow
* Warning: there is 2 types of workflows, the resource workflow and the native workflow
* native workflow is the one that you create to schedule an execution
* resource workflow is the one that is created to set our native workflow in catalog
 */
type AbstractWorkflow struct {
	resources.ResourceSet
	Graph          *graph.Graph      `bson:"graph,omitempty" json:"graph,omitempty"`       // Graph UI & logic representation of the workflow
	ScheduleActive bool              `json:"schedule_active" bson:"schedule_active"`       // ScheduleActive is a flag that indicates if the schedule is active, if not the workflow is not scheduled and no execution or booking will be set
	Schedule       *WorkflowSchedule `bson:"schedule,omitempty" json:"schedule,omitempty"` // Schedule is the schedule of the workflow
	Shared         []string          `json:"shared,omitempty" bson:"shared,omitempty"`     // Shared is the ID of the shared workflow
}

func (w *AbstractWorkflow) GetWorkflows() (list_computings []graph.GraphItem) {
	for _, item := range w.Graph.Items {
		if item.Workflow != nil {
			list_computings = append(list_computings, item)
		}
	}
	return
}

func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []*compute.ComputeResource {
	storages := []*compute.ComputeResource{}
	for _, link := range w.Graph.Links {
		nodeID := link.Destination.ID                 // we considers that the processing is the destination
		node := w.Graph.Items[link.Source.ID].Compute // we are looking for the storage as source
		if node == nil {                              // if the source is not a storage, we consider that the destination is the storage
			nodeID = link.Source.ID                           // and the processing is the source
			node = w.Graph.Items[link.Destination.ID].Compute // we are looking for the storage as destination
		}
		if processingID == nodeID && node != nil { // if the storage is linked to the processing
			storages = append(storages, node)
		}
	}
	return storages
}

func (w *AbstractWorkflow) GetStoragesByRelatedProcessing(processingID string) []*storage.StorageResource {
	storages := []*storage.StorageResource{}
	for _, link := range w.Graph.Links {
		nodeID := link.Destination.ID                 // we considers that the processing is the destination
		node := w.Graph.Items[link.Source.ID].Storage // we are looking for the storage as source
		if node == nil {                              // if the source is not a storage, we consider that the destination is the storage
			nodeID = link.Source.ID                           // and the processing is the source
			node = w.Graph.Items[link.Destination.ID].Storage // we are looking for the storage as destination
		}
		if processingID == nodeID && node != nil { // if the storage is linked to the processing
			storages = append(storages, node)
		}
	}
	return storages
}

func (w *AbstractWorkflow) GetProcessings() (list_computings []graph.GraphItem) {
	for _, item := range w.Graph.Items {
		if item.Processing != nil {
			list_computings = append(list_computings, item)
		}
	}
	return
}

// tool function to check if a link is a link between a compute and a resource
func (w *AbstractWorkflow) isDCLink(link graph.GraphLink) (bool, string) {
	if w.Graph == nil || w.Graph.Items == nil {
		return false, ""
	}
	if d, ok := w.Graph.Items[link.Source.ID]; ok && d.Compute != nil {
		return true, d.Compute.UUID
	}
	if d, ok := w.Graph.Items[link.Destination.ID]; ok && d.Compute != nil {
		return true, d.Compute.UUID
	}
	return false, ""
}

/*
* Workflow is a struct that represents a workflow
* it defines the native workflow
 */
type Workflow struct {
	utils.AbstractObject // AbstractObject contains the basic fields of an object (id, name)
	AbstractWorkflow     // AbstractWorkflow contains the basic fields of a workflow
}

/*
* CheckBooking is a function that checks the booking of the workflow on peers (even ourselves)
 */
func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
	// check if
	if wfa.Graph == nil { // no graph no booking
		return false, nil
	}
	accessor := (&compute.ComputeResource{}).GetAccessor("", []string{}, caller)
	for _, link := range wfa.Graph.Links {
		if ok, dc_id := wfa.isDCLink(link); ok { // check if the link is a link between a compute and a resource
			dc, code, _ := accessor.LoadOne(dc_id)
			if code != 200 {
				continue
			}
			// CHECK BOOKING ON PEER, compute could be a remote one
			peerID := dc.(*compute.ComputeResource).PeerID
			if peerID == "" {
				return false, errors.New("no peer id")
			} // no peer id no booking, we need to know where to book
			_, err := (&peer.Peer{}).LaunchPeerExecution(peerID, dc_id, tools.BOOKING, tools.GET, nil, caller)
			if err != nil {
				return false, err
			}
		}
	}
	return true, nil
}

func (d *Workflow) GetAccessor(peerID string, groups []string, caller *tools.HTTPCaller) utils.Accessor {
	return New(tools.WORKFLOW, peerID, groups, caller) // Create a new instance of the accessor
}