manager

type AppSessionEventHandler ¶

type AppSessionEventHandler struct {
	// contains filtered or unexported fields
}

AppSessionEventHandler listens for ContainerEvent on the event bus and synchronizes the corresponding AppSession status accordingly.

When a container transitions to creating/running/stopped/failed, this handler updates the owning AppSession's Status, StartedAt, and StoppedAt fields so that the AppSession reflects the actual container lifecycle.

Only containers with OwnerType == ContainerOwnerAppSession are processed; DAG nodes and services are ignored.

func NewAppSessionEventHandler(repo interfaces.ContainerRepository, hub *realtime.Hub) *AppSessionEventHandler

func (h *AppSessionEventHandler) Handle(evt event.Event)

type ContainerCreateWorker ¶

type ContainerCreateWorker struct {
	// contains filtered or unexported fields
}

ContainerCreateWorker subscribes to OutboxCreateRequestEvent and ContainerEvent from the event bus. For creation requests it executes rt.Create + rt.Start, and updates the active create request count while the request is being handled.

func NewContainerCreateWorker(
	repo interfaces.ContainerRepository,
	projectRepo interfaces.ProjectRepository,
	analysisRepo interfaces.AnalysisRepository,
	workflowService interfaces.WorkflowService,
	reg *containerruntime.Registry,
	res ContainerRuntimeResolver,

	containerManager *ContainerManager,
	cfg *config.Config,
) *ContainerCreateWorker

func (w *ContainerCreateWorker) Handle(evt event.Event)

type ContainerManager ¶

type ContainerManager struct {
	// contains filtered or unexported fields
}

func NewContainerManager(
	containerRepo interfaces.ContainerRepository,
	analysisRepo interfaces.AnalysisRepository,
	projectRepo interfaces.ProjectRepository,
	workflowService interfaces.WorkflowService,
	reg *containerruntime.Registry,
	bus event.Bus,
	res ContainerRuntimeResolver,
	img *ImageManager,
	cfg *config.Config,
) *ContainerManager

func (m *ContainerManager) CreateByTemplate(
	ctx context.Context,
	templateID int64,
	ownerType types.ContainerOwnerType,
	ownerID int64,
	name string,
) (*types.ContainerInstance, error)

	return nil
}

func (m *ContainerManager) CreateQueueMaxPending() int

func (m *ContainerManager) Delete(ctx context.Context, id int64) error

func (m *ContainerManager) GetLogs(ctx context.Context, id int64, tail int) (string, error)

func (s *ContainerManager) GetMaxConcurrency() int

func (m *ContainerManager) OnEvent(e containerruntime.RuntimeEvent)

func (m *ContainerManager) QueueStatus(ctx context.Context) (*CreateQueueStatus, error)

func (m *ContainerManager) RecoverRuntimeMonitoring(ctx context.Context) (int, error)

func (m *ContainerManager) RunRuntimeReconciler(ctx context.Context, interval time.Duration)

func (m *ContainerManager) Start(ctx context.Context, id int64) error

func (m *ContainerManager) Stop(ctx context.Context, id int64) error

func (m *ContainerManager) StopByOwner(ctx context.Context, ownerType types.ContainerOwnerType, ownerID int64) error

func (s *ContainerManager) TransitionContainerAndEnqueueOutbox(ctx context.Context, inst *types.ContainerInstance, to types.ContainerStatus, eventType string) error

type ContainerRuntimeResolveInput ¶

type ContainerRuntimeResolveInput struct {
	Spec      *types.ContainerSpec
	Variables map[string]string
}

type ContainerRuntimeResolver ¶

type ContainerRuntimeResolver interface {
	Resolve(ctx context.Context, in *ContainerRuntimeResolveInput) (*types.ContainerSpec, error)
}

func NewDefaultContainerRuntimeResolver() ContainerRuntimeResolver

type CreateQueueStatus ¶

type CreateQueueStatus struct {
	ActiveCount    int64
	PendingCount   int64
	MaxConcurrency int
	MaxPending     int
}

type ImageManager ¶

type ImageManager struct {
	// contains filtered or unexported fields
}

func NewImageManager(repo interfaces.ContainerRepository, reg *containerruntime.Registry) *ImageManager

type OutboxCreateRequestEvent ¶

type OutboxCreateRequestEvent struct {
	OutboxID   int64
	RawPayload []byte
}

OutboxCreateRequestEvent is published to the event bus when a ContainerCreateRequest outbox event is picked up. Handlers that need to process deferred container creation subscribe to this.

type OutboxDeleteRequestEvent ¶

type OutboxDeleteRequestEvent struct {
	OutboxID   int64
	RawPayload []byte
}

OutboxDeleteRequestEvent is published to the event bus when a ContainerDeleteRequest outbox event is picked up.

type OutboxDispatcher ¶

type OutboxDispatcher struct {
	// contains filtered or unexported fields
}

func NewOutboxDispatcher(repo interfaces.ContainerRepository, bus event.Bus) *OutboxDispatcher

func (d *OutboxDispatcher) Start(ctx context.Context)

type OutboxRecreateRequestEvent ¶

type OutboxRecreateRequestEvent struct {
	OutboxID   int64
	RawPayload []byte
}

type OutboxStartRequestEvent ¶

type OutboxStartRequestEvent struct {
	OutboxID   int64
	RawPayload []byte
}

OutboxStartRequestEvent is published to the event bus when a ContainerStartRequest outbox event is picked up.

type OutboxStopRequestEvent ¶

type OutboxStopRequestEvent struct {
	OutboxID   int64
	RawPayload []byte
}

OutboxStopRequestEvent is published to the event bus when a ContainerStopRequest outbox event is picked up.