tools

package
v0.0.0-...-cef23b5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 27, 2026 License: AGPL-3.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var CATALOGAPI = func() string {
	return config.GetConfig().InternalCatalogAPI
}
View Source 
var DATACENTERAPI = func() string {
	return config.GetConfig().InternalDatacenterAPI
}
View Source 
var InnerDefaultAPI = [...]func() string{
	NOAPI,
	CATALOGAPI,
	CATALOGAPI,
	CATALOGAPI,
	CATALOGAPI,
	CATALOGAPI,
	WORKFLOWAPI,
	NOAPI,
	WORKSPACEAPI,
	PEERSAPI,
	SHAREDAPI,
	SHAREDAPI,
	SCHEDULERAPI,
	NOAPI,
	NOAPI,
	NOAPI,
	PURCHASEAPI,
	DATACENTERAPI,
	DATACENTERAPI,
	NOAPI,
	CATALOGAPI,
	SCHEDULERAPI,
	DATACENTERAPI,
	CATALOGAPI,
	CATALOGAPI,
	DATACENTERAPI,
	NOAPI,
	NOAPI,
	NOAPI,
	NOAPI,
}

Bind the standard API name to the data type

View Source 
var NOAPI = func() string {
	return ""
}
View Source 
var PEERSAPI = func() string {
	return config.GetConfig().InternalPeerAPI
}
View Source 
var PURCHASEAPI = func() string {
	return config.GetConfig().InternalCatalogAPI + "/purchase"
}
View Source 
var SCHEDULERAPI = func() string {
	return config.GetConfig().InternalSchedulerAPI
}
View Source 
var SHAREDAPI = func() string {
	return config.GetConfig().InternalSharedAPI
}
View Source 
var Str = [...]string{
	"invalid",
	"data_resource",
	"processing_resource",
	"storage_resource",
	"compute_resource",
	"workflow_resource",
	"workflow",
	"workflow_execution",
	"workspace",
	"peer",
	"collaborative_area",
	"rule",
	"booking",
	"workflow_history",
	"workspace_history",
	"order",
	"purchase_resource",
	"live_datacenter",
	"live_storage",
	"bill",
	"native_tool",
	"execution_verification",
	"allowed_image",
	"service_resource",
	"dynamic_resource",
	"live_service",
	"payment",
	"refund",
	"discount",
	"subscription",
}

Bind the standard data name to the data type

View Source 
var UncatchedError = []error{} // Singleton instance of the api 500 error cache

* API is the Health Check API * it defines the health check methods

View Source 
var WORKFLOWAPI = func() string {
	return config.GetConfig().InternalWorkflowAPI
}
View Source 
var WORKSPACEAPI = func() string {
	return config.GetConfig().InternalWorkspaceAPI
}

Functions

func FromInt 

func FromInt(i int) string

func FromString 

func FromString(comp string) int

func GetConcatenatedName 

func GetConcatenatedName(peerId string, namespace string) string

============== ADMIRALTY ============== Returns a concatenation of the peerId and namespace in order for kubernetes ressources to have a unique name, under 63 characters and yet identify which peer they are created for

func LoadKeyFromFilePrivate 

func LoadKeyFromFilePrivate() (crypto.PrivKey, error)

func LoadKeyFromFilePublic 

func LoadKeyFromFilePublic() (crypto.PubKey, error)

func NewDynamicClient 

func NewDynamicClient(host string, ca string, cert string, data string) (*dynamic.DynamicClient, error)

func NewNATSCaller 

func NewNATSCaller() *natsCaller

NewNATSCaller creates a new instance of the NATS Caller

Types

type API 

type API struct{}

func (*API) CheckRemoteAPIs 

func (a *API) CheckRemoteAPIs(apis []DataType) (State, map[string]string, error)

CheckRemoteAPIs checks the state of remote APIs from your proper OC

func (*API) CheckRemotePeer 

func (a *API) CheckRemotePeer(url string) (State, map[string]int)

CheckRemotePeer checks the state of a remote peer

func (*API) Discovered 

func (a *API) Discovered(infos []*beego.ControllerInfo, extra ...map[string][]string)

func (*API) GetState 

func (a *API) GetState() (State, int, error)

GetState returns the state of the API

func (*API) ListenRouter 

func (a *API) ListenRouter(exec func(msg NATSResponse))

func (*API) SubscribeRouter 

func (a *API) SubscribeRouter(infos []*beego.ControllerInfo, extra ...map[string][]string)

type APIRequest 

type APIRequest struct {
	Username string
	PeerID   string
	Groups   []string
	Caller   *HTTPCaller
	Admin    bool
}

type APIStatus 

type APIStatus struct {
	Code     int               `json:"code"`     // Code is the status code, where 0 is ALIVE, 1 is REDUCED_SERVICE, 2 is UNPROCESSABLE_ENTITY, 3 is DB_FALLOUT, 4 is TEAPOT, 5 is DEAD
	State    string            `json:"state"`    // State is the state of the API (status shows as a string) (alive, reduced service, unprocessable entity, database fallout, some things boils in here, i'm probably a teapot, dead)
	Services map[string]string `json:"services"` // Services is the state of the services of the API (status shows as a string) (alive, reduced service, unprocessable entity, database fallout, some things boils in here, i'm probably a teapot, dead)
}

* APIStatus is the status of the API * it defines the state of the API * Code is the status code, where 0 is ALIVE, 1 is REDUCED_SERVICE, 2 is UNPROCESSABLE_ENTITY, 3 is DB_FALLOUT, 4 is TEAPOT, 5 is DEAD

type APIStatusResponse 

type APIStatusResponse struct {
	Data  *APIStatus `json:"data"`
	Error string     `json:"error"`
}

APIStatusResponse is the response of the API status

type BehaviorSeverity 

type BehaviorSeverity int

BehaviorSeverity qualifies the gravity of a peer misbehavior. 

const (
	// BehaviorWarn: minor inconsistency — slight trust penalty.
	BehaviorWarn BehaviorSeverity = iota
	// BehaviorFraud: deliberate data manipulation (e.g. fake peerless Ref,
	// invalid booking) — significant trust penalty.
	BehaviorFraud
	// BehaviorCritical: severe abuse (secret exfiltration, data corruption,
	// system-level attack) — heavy penalty, near-immediate blacklist.
	BehaviorCritical
)

func (BehaviorSeverity) Penalty 

func (s BehaviorSeverity) Penalty() float64

Penalty returns the trust-score deduction for this severity.

type DataType 

type DataType int
const (
	INVALID DataType = iota
	DATA_RESOURCE
	PROCESSING_RESOURCE
	STORAGE_RESOURCE
	COMPUTE_RESOURCE
	WORKFLOW_RESOURCE
	WORKFLOW
	WORKFLOW_EXECUTION
	WORKSPACE
	PEER
	COLLABORATIVE_AREA
	RULE
	BOOKING
	WORKFLOW_HISTORY
	WORKSPACE_HISTORY
	ORDER
	PURCHASE_RESOURCE
	LIVE_DATACENTER
	LIVE_STORAGE
	BILL
	NATIVE_TOOL
	EXECUTION_VERIFICATION
	ALLOWED_IMAGE
	SERVICE_RESOURCE
	DYNAMIC_RESOURCE
	LIVE_SERVICE
	PAYMENT
	REFUND
	DISCOUNT
	SUBSCRIPTION
)

DataType - Enum for the different types of resources in db accessible from the outside

func DataTypeList 

func DataTypeList() []DataType

func (DataType) EnumIndex 

func (d DataType) EnumIndex() int

EnumIndex - Creating common behavior - give the type a EnumIndex functio

func (DataType) InnerAPI 

func (d DataType) InnerAPI() string

func (DataType) String 

func (d DataType) String() string

type HTTPCaller 

type HTTPCaller struct {
	URLS        map[DataType]map[METHOD]string // Map of the different methods and their urls
	Disabled    bool                           // Disabled flag
	Mu          sync.RWMutex
	LastResults map[string]interface{} // Used to store information regarding the last execution of a given method on a given data type
}

func NewHTTPCaller 

func NewHTTPCaller(urls map[DataType]map[METHOD]string) *HTTPCaller

NewHTTPCaller creates a new instance of the HTTP Caller

func (*HTTPCaller) CallDelete 

func (caller *HTTPCaller) CallDelete(url string, subpath string) ([]byte, error)

CallPut calls the DELETE method on the HTTP server

func (*HTTPCaller) CallForm 

func (caller *HTTPCaller) CallForm(method string, url string, subpath string,
	body url.Values, content_type string, fakeTLSTermination bool, cookies ...*http.Cookie) (*http.Response, error)

CallRaw calls the Raw method on the HTTP server

func (*HTTPCaller) CallGet 

func (caller *HTTPCaller) CallGet(url string, subpath string, types ...string) ([]byte, error)

CallGet calls the GET method on the HTTP server

func (*HTTPCaller) CallPost 

func (caller *HTTPCaller) CallPost(url string, subpath string, body interface{}, types ...string) ([]byte, error)

CallPost calls the POST method on the HTTP server

func (*HTTPCaller) CallPut 

func (caller *HTTPCaller) CallPut(url string, subpath string, body map[string]interface{}) ([]byte, error)

CallPost calls the POST method on the HTTP server

func (*HTTPCaller) CallRaw 

func (caller *HTTPCaller) CallRaw(method string, url string, subpath string,
	body map[string]interface{}, content_type string, fakeTLSTermination bool, cookies ...*http.Cookie) (*http.Response, error)

CallRaw calls the Raw method on the HTTP server

func (*HTTPCaller) DeepCopy 

func (c *HTTPCaller) DeepCopy(dst HTTPCaller) error

Creates a copy of the current caller, in order to have parallelized executions without race condition

func (*HTTPCaller) GetUrls 

func (c *HTTPCaller) GetUrls() map[DataType]map[METHOD]string

func (*HTTPCaller) StoreResp 

func (caller *HTTPCaller) StoreResp(resp *http.Response) error

type HTTPCallerITF 

type HTTPCallerITF interface {
	GetUrls() map[DataType]map[METHOD]string
	CallGet(url string, subpath string, types ...string) ([]byte, error)
	CallPost(url string, subpath string, body interface{}, types ...string) ([]byte, error)
	CallDelete(url string, subpath string) ([]byte, error)
}

type KubernetesService 

type KubernetesService struct {
	Set  *kubernetes.Clientset
	Host string
	CA   string
	Cert string
	Data string
}

func NewKubernetesService 

func NewKubernetesService(host string, ca string, cert string, data string) (*KubernetesService, error)

func (*KubernetesService) CheckHealth 

func (k *KubernetesService) CheckHealth() error

func (*KubernetesService) CreateAdmiraltySource 

func (k *KubernetesService) CreateAdmiraltySource(context context.Context, executionId string) ([]byte, error)

Admiralty Source allows a cluster to receive pods from a remote cluster

The source must be associated to a serviceAccount, which will execute the pods locally. This serviceAccount must have sufficient permission to create and patch pods

This method is temporary to implement the use of Admiralty, but must be edited to rather contact the oc-datacenter from the remote cluster to create the source locally and retrieve the token for the serviceAccount

func (*KubernetesService) CreateAdmiraltyTarget 

func (k *KubernetesService) CreateAdmiraltyTarget(context context.Context, executionId string, peerId string) ([]byte, error)

Admiralty Target allows a cluster to deploy pods to remote cluster

The remote cluster must :

- have declared a Source resource

- have declared the same namespace as the one where the pods are created in the local cluster

- have delcared a serviceAccount with sufficient permission to create pods

func (*KubernetesService) CreateKubeconfigSecret 

func (k *KubernetesService) CreateKubeconfigSecret(context context.Context, kubeconfig string, executionId string, peerId string) ([]byte, error)

Create a secret from a kubeconfing. Use it to create the secret binded to an Admiralty target, which must contain the serviceAccount's token value

func (*KubernetesService) CreateNamespace 

func (k *KubernetesService) CreateNamespace(ctx context.Context, ns string) error

func (*KubernetesService) CreatePVC 

func (k *KubernetesService) CreatePVC(ctx context.Context, name, namespace, storageSize string) error

CreatePVC creates a static PersistentVolume + PersistentVolumeClaim in the given namespace. Static provisioning (no StorageClass) avoids the WaitForFirstConsumer deadlock with Admiralty virtual nodes — the PVC binds immediately.

func (*KubernetesService) CreateRole 

func (k *KubernetesService) CreateRole(ctx context.Context, ns string, role string, groups [][]string, resources [][]string, verbs [][]string) error

func (*KubernetesService) CreateRoleBinding 

func (k *KubernetesService) CreateRoleBinding(ctx context.Context, ns string, roleBinding string, role string) error

func (*KubernetesService) CreateSecret 

func (k *KubernetesService) CreateSecret(context context.Context, minioId string, executionID string, access string, secret string) error

func (*KubernetesService) CreateServiceAccount 

func (k *KubernetesService) CreateServiceAccount(ctx context.Context, ns string) error

func (*KubernetesService) DeleteKubeConfigSecret 

func (k *KubernetesService) DeleteKubeConfigSecret(executionID string) ([]byte, error)

func (*KubernetesService) DeleteNamespace 

func (k *KubernetesService) DeleteNamespace(ctx context.Context, ns string, f func()) error

func (*KubernetesService) DeletePVC 

func (k *KubernetesService) DeletePVC(ctx context.Context, name, namespace string) error

DeletePVC deletes a PersistentVolumeClaim and its associated PersistentVolume.

func (*KubernetesService) DeleteSecretsByLabel 

func (k *KubernetesService) DeleteSecretsByLabel(ctx context.Context, namespace, labelSelector string) error

DeleteSecretsByLabel deletes all Secrets in the given namespace matching labelSelector. Used by oc-datacenter to clean up ephemeral source-presigned Secrets after workflow completion.

func (*KubernetesService) GenerateToken 

func (k *KubernetesService) GenerateToken(ctx context.Context, ns string, duration int) (string, error)

Returns the string representing the token generated for the serviceAccount in the namespace identified by the value `ns` with the name sa-`ns`, which is valid for `duration` seconds

func (*KubernetesService) GetKubeconfigSecret 

func (k *KubernetesService) GetKubeconfigSecret(context context.Context, executionId string, peerId string) ([]byte, error)

func (*KubernetesService) GetNamespace 

func (k *KubernetesService) GetNamespace(context context.Context, executionID string) (*v1.Namespace, error)

func (*KubernetesService) GetOneNode 

func (k *KubernetesService) GetOneNode(context context.Context, executionID string, peerId string) (*v1.Node, error)

Returns the Kubernetes' Node object corresponding to the executionID if it exists on this host

The node is created when an admiralty Target (on host) can connect to an admiralty Source (on remote)

func (*KubernetesService) GetTargets 

func (k *KubernetesService) GetTargets(ctx context.Context) ([]string, error)

Needs refactoring :

  • Retrieving the metada (in a method that Unmarshall the part of the json in a metadata object)

func (*KubernetesService) ProvisionExecutionNamespace 

func (k *KubernetesService) ProvisionExecutionNamespace(ctx context.Context, ns string) error

ProvisionExecutionNamespace creates the full Argo execution environment for a namespace: namespace, service-account, role and role-binding. Idempotent — if the namespace already exists the call is a no-op.

func (*KubernetesService) TeardownExecutionNamespace 

func (k *KubernetesService) TeardownExecutionNamespace(ctx context.Context, ns string) error

TeardownExecutionNamespace deletes the namespace and lets Kubernetes cascade the deletion of all contained resources (SA, Role, RoleBinding, pods…).

type METHOD 

type METHOD int

HTTP Method Enum defines the different methods that can be used to interact with the HTTP server 

const (
	GET METHOD = iota
	PUT
	POST
	POSTCHECK
	DELETE

	STRICT_INTERNAL_GET
	STRICT_INTERNAL_PUT
	STRICT_INTERNAL_POST
	STRICT_INTERNAL_DELETE
)

func ToMethod 

func ToMethod(str string) METHOD

ToMethod returns the method from a string

func (METHOD) EnumIndex 

func (m METHOD) EnumIndex() int

EnumIndex returns the index of the enum

func (METHOD) String 

func (m METHOD) String() string

String returns the string of the enum

type NATSMethod 

type NATSMethod int

NATS Method Enum defines the different methods that can be used to interact with the NATS server 

const (
	REMOVE_EXECUTION NATSMethod = iota
	CREATE_EXECUTION
	PLANNER_EXECUTION
	DISCOVERY

	WORKFLOW_EVENT
	ARGO_KUBE_EVENT

	CREATE_RESOURCE
	REMOVE_RESOURCE

	PROPALGATION_EVENT
	SEARCH_EVENT

	CONFIRM_EVENT

	CONSIDERS_EVENT
	ADMIRALTY_CONFIG_EVENT
	MINIO_CONFIG_EVENT
	PVC_CONFIG_EVENT

	// Workflow lifecycle events emitted by oc-monitord.
	// oc-scheduler listens to STARTED and DONE to maintain WorkflowExecution state.
	// oc-datacenter listens to STEP_DONE and DONE to close bookings and tear down infra.
	WORKFLOW_STARTED_EVENT
	WORKFLOW_STEP_DONE_EVENT
	WORKFLOW_DONE_EVENT

	// PEER_BEHAVIOR_EVENT is emitted by any trusted service (oc-scheduler,
	// oc-datacenter, …) when a peer exhibits suspicious or fraudulent behavior.
	// oc-discovery consumes it to update the peer's trust score and auto-blacklist
	// below threshold.
	PEER_BEHAVIOR_EVENT

	// PEER_OBSERVE_RESPONSE_EVENT is emitted by oc-discovery each time it receives
	// a heartbeat from an observed remote peer. oc-peer listens to this event to
	// update the WS connectivity state for its clients.
	PEER_OBSERVE_RESPONSE_EVENT

	// PEER_OBSERVE_EVENT is emitted by oc-peer to request oc-discovery to start
	// or stop observing a remote peer. Payload contains the target peer_id and a
	// boolean close flag.
	PEER_OBSERVE_EVENT

	// SOURCE_PRESIGN_EVENT is emitted by oc-discovery on the resource-owner peer
	// when it receives a PB_SOURCE_PRESIGN request routed via libp2p.
	// oc-datacenter listens to it to generate a pre-signed Minio URL and reply
	// via PB_CONSIDERS (Phase 4 — isReachable=false).
	SOURCE_PRESIGN_EVENT
)

func NameToMethod 

func NameToMethod(name string) NATSMethod

NameToMethod returns the NATSMethod enum value from a string

func (NATSMethod) GenerateKey 

func (d NATSMethod) GenerateKey() string

GenerateKey generates a key for the NATSMethod usefull for standard key based on data name & method

func (NATSMethod) String 

func (n NATSMethod) String() string

type NATSResponse 

type NATSResponse struct {
	FromApp    string   `json:"from_app"`
	Datatype   DataType `json:"datatype"`
	User       string   `json:"user"`
	Groups     []string `json:"groups"`
	Method     int      `json:"method"`
	SearchAttr string   `json:"search_attr"`
	Payload    []byte   `json:"payload"`
}

type PeerBehaviorReport 

type PeerBehaviorReport struct {
	// ReporterApp identifies the emitting service (e.g. "oc-scheduler", "oc-datacenter").
	ReporterApp string `json:"reporter_app"`
	// TargetPeerID is the MongoDB DID (_id) of the offending peer.
	TargetPeerID string `json:"target_peer_id"`
	// Severity drives how much the trust score drops.
	Severity BehaviorSeverity `json:"severity"`
	// Reason is a human-readable description shown in the blacklist warning.
	Reason string `json:"reason"`
	// Evidence is an optional reference (booking ID, resource Ref, …).
	Evidence string `json:"evidence,omitempty"`
	// At is the timestamp of the observed misbehavior.
	At time.Time `json:"at"`
}

PeerBehaviorReport is the payload carried by PEER_BEHAVIOR_EVENT. Any trusted service can emit it; oc-discovery is the sole consumer.

type PropalgationMessage 

type PropalgationMessage struct {
	DataType int          `json:"datatype"`
	Action   PubSubAction `json:"action"`
	Payload  []byte       `json:"payload"`
}

type PubSubAction 

type PubSubAction int
const (
	PB_SEARCH PubSubAction = iota
	PB_SEARCH_RESPONSE
	PB_CREATE
	PB_UPDATE
	PB_DELETE
	PB_PLANNER
	PB_CLOSE_PLANNER
	PB_CONSIDERS
	PB_ADMIRALTY_CONFIG
	PB_MINIO_CONFIG
	PB_PVC_CONFIG
	PB_CLOSE_SEARCH
	NONE
	PB_OBSERVE
	PB_OBSERVE_CLOSE
	// PB_PROPAGATE is used by oc-discovery to broadcast a peer's online/offline
	// state to other oc-discovery nodes in the federation via PROPALGATION_EVENT.
	PB_PROPAGATE
	// PB_SOURCE_PRESIGN is sent by oc-datacenter to request a pre-signed Minio URL
	// for a private source resource (isReachable=false, Phase 4).
	// oc-discovery routes it to the resource owner peer via ProtocolSourcePresignResource.
	PB_SOURCE_PRESIGN
)

func GetActionString 

func GetActionString(ss string) PubSubAction

func (PubSubAction) String 

func (m PubSubAction) String() string

type State 

type State int
const (
	ALIVE                State = iota
	REDUCED_SERVICE            // occurs when some services are down
	UNPROCESSABLE_ENTITY       // occurs when the database is up but the collections are not
	DB_FALLOUT                 // occurs when the database is down
	TEAPOT                     // well some things boils in here, i'm probably a teapot, occurs when uncatched errors are present (it's fun)
	DEAD                       // occurs when the peer is dead
)

State is an enum that defines the state of the API

func ToState 

func ToState(str string) State

ToState returns the state from a string

func (State) EnumIndex 

func (s State) EnumIndex() int

EnumIndex returns the index of the enum

func (State) String 

func (s State) String() string

String returns the string of the enum

type StepMetric 

type StepMetric struct {
	BookingID string     `json:"booking_id"`
	State     int        `json:"state"`
	RealStart *time.Time `json:"real_start,omitempty"`
	RealEnd   *time.Time `json:"real_end,omitempty"`
}

StepMetric carries the outcome of one Argo step node as observed by oc-monitord. Embedded in WorkflowLifecycleEvent.Steps for the WORKFLOW_DONE_EVENT recap.

type WorkflowLifecycleEvent 

type WorkflowLifecycleEvent struct {
	ExecutionID  string       `json:"execution_id"`
	ExecutionsID string       `json:"executions_id"`
	BookingID    string       `json:"booking_id,omitempty"`
	State        int          `json:"state"`
	RealStart    *time.Time   `json:"real_start,omitempty"`
	RealEnd      *time.Time   `json:"real_end,omitempty"`
	Steps        []StepMetric `json:"steps,omitempty"`
}

WorkflowLifecycleEvent is the NATS payload emitted by oc-monitord on WORKFLOW_STARTED_EVENT, WORKFLOW_STEP_DONE_EVENT, and WORKFLOW_DONE_EVENT.

  • ExecutionID : WorkflowExecution UUID (used by oc-scheduler to update state)
  • ExecutionsID : run-group ID shared by all bookings of the same run
  • BookingID : non-empty only for WORKFLOW_STEP_DONE_EVENT
  • State : target state (enum index: SUCCESS=3, FAILURE=4, STARTED=2, …)
  • RealStart : actual start timestamp recorded by Argo (nil if unknown)
  • RealEnd : actual end timestamp recorded by Argo (nil for STARTED events)
  • Steps : non-nil only for WORKFLOW_DONE_EVENT — full recap of every step so oc-scheduler and oc-catalog can catch up if they missed STEP_DONE events

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.