core

README

Open Integration - a pipeline execution engine

Til the project has not reached version > 1.x.x it may have breaking changes in the API, please use the latest version before opening issue.

Concepts

• A compiled, binray pipeline
• State - the engine holds the state of all the tasks
• Service - a standalone binary exposing API over http2 (gRPC) that the engine can trigger, called endpoint.
  • Endpoint of a service defined by 2 files of JSON schema, arguments.json and returns.json, the engine will enforce the arguments given by a task and the output created to match the schema.
• Task - execution flow with input and output.
  • Service Task will send a request to service based on the endpoint.

Architecture

Dataflow

Example

use oictl to generate hello-world pipeline

• ioctl generate pipeline
• go run *.go

Real world examples

• JIRA
• Trello
• open-integration/core-ci-pipeline

Documentation

Index

• func ConditionCombined(conditions ...condition.Condition) condition.Condition
• func ConditionEngineStarted() condition.Condition
• func ConditionTaskEventReported(name string) condition.Condition
• func ConditionTaskFinished(task string) condition.Condition
• func ConditionTaskFinishedWithStatus(task string, status string) condition.Condition
• func HandleEngineError(err error)
• func NewFunctionTask(name string, fn func(context.Context, task.RunOptions) ([]byte, error)) task.Task
• func NewSerivceTask(name string, service string, endpoint string, arg ...task.Argument) task.Task
• func NewTickerTask(name string, tickInterval time.Duration, totalTime time.Duration) task.Task
• type Engine
  • func NewEngine(opt *EngineOptions) Engine
• type EngineKubernetesOptions
• type EngineOptions
• type EventReaction
• type Pipeline
• type PipelineMetadata
• type PipelineSpec
• type Service

Functions

func ConditionCombined

func ConditionCombined(conditions ...condition.Condition) condition.Condition

ConditionCombined returns the condition that is evaluated to true when all the conditions are true

func ConditionEngineStarted

func ConditionEngineStarted() condition.Condition

ConditionEngineStarted returns the condition that is evaluated to true on engine.started event

func ConditionTaskEventReported

func ConditionTaskEventReported(name string) condition.Condition

ConditionTaskEventReported return the condition that satisfied when task reported event in format {TASK_NAME}.{EVENT}

func ConditionTaskFinished

func ConditionTaskFinished(task string) condition.Condition

ConditionTaskFinished returns the condition that is evaluated to true on task.finished event and the task is marked as finished in the state

func ConditionTaskFinishedWithStatus

func ConditionTaskFinishedWithStatus(task string, status string) condition.Condition

ConditionTaskFinishedWithStatus returns the condition that is evaluated to true on task.finished event and the task is marked as finished in the state and the status is as given

func HandleEngineError

func HandleEngineError(err error)

HandleEngineError prints the error in case the engine.Run was failed and exit

func NewFunctionTask

func NewFunctionTask(name string, fn func(context.Context, task.RunOptions) ([]byte, error)) task.Task

NewFunctionTask build task that will be executed in same process

func NewSerivceTask

func NewSerivceTask(name string, service string, endpoint string, arg ...task.Argument) task.Task

NewSerivceTask build task task calls a service with arguments

func NewTickerTask

func NewTickerTask(name string, tickInterval time.Duration, totalTime time.Duration) task.Task

NewTickerTask builds task that will send event every tickInterval till it stops on totalTime

Types

type Engine

type Engine interface {
	Run() error
	Modem() modem.Modem
}

Engine exposes the interface of the engine

func NewEngine

func NewEngine(opt *EngineOptions) Engine

NewEngine create new engine

type EngineKubernetesOptions

type EngineKubernetesOptions struct {
	Path                string
	Context             string
	Namespace           string
	InCluster           bool
	Host                string
	B64Crt              string
	Token               string
	LogsVolumeClaimName string
	LogsVolumeName      string
}

EngineKubernetesOptions when running service on kubernetes cluster

type EngineOptions

type EngineOptions struct {
	Pipeline Pipeline
	// LogsDirectory path where to store logs
	LogsDirectory string
	Kubeconfig    *EngineKubernetesOptions
	Logger        logger.Logger
	// contains filtered or unexported fields
}

EngineOptions to create new engine

type EventReaction

type EventReaction struct {
	Condition condition.Condition
	Reaction  func(ev event.Event, state state.State) []task.Task
}

EventReaction is a binding of an event to a function that builds tasks

type Pipeline

type Pipeline struct {
	Metadata PipelineMetadata
	Spec     PipelineSpec
}

Pipeline is the pipeline representation

type PipelineMetadata

type PipelineMetadata struct {
	Name         string
	OS           string
	Architecture string
}

PipelineMetadata holds all the metadata of a pipeline

type PipelineSpec

type PipelineSpec struct {
	Reactions []EventReaction
	Services  []Service
}

PipelineSpec is the spec of a pipeline

type Service

type Service struct {
	// Name is official service which is part of the service catalog (https://github.com/open-integration/core-services/releases)
	Name string
	// Version of the service, empty string will use the latest version from catalog
	Version string
	// Path a location of the local fs the service can be found, Path cannot be set with Name together
	Path string
	// As alias name to refer the service as part of the task implementation
	As string
}

Service is a Service a pipeline should execute