controlapi

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Sep 30, 2026 License: Apache-2.0 Imports: 66 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrNoAteletOnNode = errors.New("no atelet pod found on node")

ErrNoAteletOnNode reports that the informer cache holds no atelet pod for the requested node — e.g. the atelet is restarting, or the node is gone. Retryable.

Functions

func AteletInformer 

func AteletInformer(kc kubernetes.Interface, ateletNamespace string) (informers.SharedInformerFactory, cache.SharedIndexInformer)

AteletInformer creates a SharedInformerFactory and SharedIndexInformer for Atelet pods in the given namespace.

func RegisterActorCrashes 

func RegisterActorCrashes(meter metric.Meter) error

RegisterActorCrashes initializes the ate.actor.crashes counter instrument.

func RegisterWorkerCount 

func RegisterWorkerCount(meter metric.Meter, workers func() ([]*ateapipb.Worker, error), listPools func(labels.Selector) ([]*atev1alpha1.WorkerPool, error)) error

RegisterWorkerCount wires the ate.workerpool.workers observable against workers (workercache.Cache.Workers in prod) and listPools (a WorkerPool lister's List, used to seed zero-valued series). Worker counts are spatially summable (over states = pool size, over pools = fleet), which is the UpDownCounter contract; a gauge would be wrong for a value meant to be summed.

Types

type ActorTemplateReconciler 

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

ActorTemplateReconciler drives stored ActorTemplates through the golden actor state machine.

func NewActorTemplateReconciler 

func NewActorTemplateReconciler(persistence templateReconcilerStore, control goldenActorControl, resyncInterval time.Duration) *ActorTemplateReconciler

func (*ActorTemplateReconciler) Start 

func (r *ActorTemplateReconciler) Start(ctx context.Context)

Start launches the queue workers and the resync producer; there is no event source for stored templates, so the periodic list is the event source.

type ActorWorkflow 

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

ActorWorkflow handles the workflows for actor's resume / suspend operations.

func NewActorWorkflow 

func NewActorWorkflow(
	store actorWorkflowStore,
	workerCache *workercache.Cache,
	dialer *AteletDialer,
	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
	storageClassLister storagev1listers.StorageClassLister,
	instruments *Instruments,
	egressGatewayAddress string,
	pluginRegistry VolumePluginRegistry,
	objectStore objectstore.Store,
) *ActorWorkflow

NewActorWorkflow creates a new ActorWorkflow. instruments may be nil.

objectStore may be nil, which leaves external snapshots in place instead of copying and releasing them. Only tests that never reach those steps pass nil; ate-api always builds one.

func (*ActorWorkflow) DeleteActor 

func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.ActorRef, anyState bool, precondition store.DeletePreconditions) (*ateapipb.Actor, error)

DeleteActor executes the workflow to delete an actor. Idempotent.

func (*ActorWorkflow) DeleteActorTemplate added in v0.2.0 

func (w *ActorWorkflow) DeleteActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, precondition store.DeletePreconditions) (*ateapipb.ActorTemplate, error)

DeleteActorTemplate executes the workflow to delete an ActorTemplate. The golden actor and tag go first and the row last, so a failed attempt leaves the row for a retry to rediscover them by.

func (*ActorWorkflow) DeleteTag added in v0.2.0 

func (w *ActorWorkflow) DeleteTag(ctx context.Context, tagRef resources.TagRef, precondition store.DeletePreconditions) (*ateapipb.Tag, error)

DeleteTag releases the external snapshot the tag owns and then removes the row, in that order: the row is the only handle on that snapshot, so dropping it first would leak.

The workflow is built in 3 phases:

  1. Load the tag (which names the snapshot to collect).
  2. Release that snapshot, tolerating a previous attempt partly collected.
  3. Finalize: drop the row.

Idempotent: a failure at any phase leaves the row in place, so the same delete run again rediscovers the work from it and resumes over whatever is left.

The tag stays resolvable while its snapshot is being collected, so a CreateActor racing this delete can seed an Actor from content that is going away. That race is accepted for now.

Note that this destroys the external snapshot: an Actor created from the tag and never suspended is still borrowing it and becomes unrecoverable. Do not delete a tag while clones of it exist.

func (*ActorWorkflow) PauseActor 

func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error)

PauseActor executes the workflow to pause a running actor. Idempotent: a re-entered workflow fast-forwards past the steps a previous attempt completed, deriving progress from the persisted actor alone.

func (*ActorWorkflow) ResumeActor 

func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, resumed bool, err error)

ResumeActor executes the workflow to resume a suspended actor. Idempotent: a re-entered workflow fast-forwards past the steps a previous attempt completed, deriving progress from the persisted actor alone.

func (*ActorWorkflow) RevertActor added in v0.2.0 

func (w *ActorWorkflow) RevertActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error)

RevertActor executes the workflow to discard an actor's current execution and return it to SUSPENDED, leaving its external snapshot untouched for a later resume to restore from. It is the recovery verb: it accepts a CRASHED actor, and needs no live worker to succeed.

Idempotent: each step keys on a persisted field rather than on the state the actor was reverted from, so a re-entered workflow fast-forwards to wherever the previous attempt stopped.

func (*ActorWorkflow) SuspendActor 

func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error)

SuspendActor executes the workflow to suspend a running or paused actor: a running actor is checkpointed on its worker, a paused actor's node-local snapshot is uploaded. Idempotent: a re-entered workflow fast-forwards past the steps a previous attempt completed, deriving progress from the persisted actor alone.

func (*ActorWorkflow) TagActorSnapshot 

func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error)

TagActorSnapshot tags the external snapshot held by the suspended actor the tag's source_actor names. The tag is given its own copy of that snapshot, so suspending the actor again or deleting the actor does not garbage collect the tag's snapshot.

The tag is built in 3 phases:

  1. Reserve the tag and record its storage location.
  2. Copy the snapshot under the reserved tag's UID.
  3. Finalize: write the completed snapshot object to the tag.

The tag captures whichever snapshot the actor holds when the workflow runs. An actor keeps no snapshot history, so a suspend that lands first moves what gets tagged; that race is inherent to naming an actor rather than a snapshot.

Not idempotent: the name is taken as soon as phase 1 lands, so a create that dies after it leaves a pending tag and every later create under that name is AlreadyExists. To retry, delete the tag, which collects whatever the failed attempt stranded, and create it again.

type AteletDialer 

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

AteletDialer handles gRPC connections to Atelet pods.

func NewAteletDialer 

func NewAteletDialer(ateletIndexer cache.Indexer, ateletSPIFFEID, clientBundlePath, serverCAPath string, opts ...DialerOption) *AteletDialer

NewAteletDialer creates a new AteletDialer. clientBundlePath and serverCAPath are used to build the per-atelet mTLS credentials used for every atelet connection, and ateletSPIFFEID is the identity those credentials expect on the atelet serving cert.

func (*AteletDialer) DialForAteletOnNode 

func (d *AteletDialer) DialForAteletOnNode(nodeName string) (*grpc.ClientConn, error)

DialForAteletOnNode resolves the single atelet pod on nodeName and dials it with per-atelet pod-UID-pinned credentials, caching the connection by the atelet's pod UID.

type DialerOption 

type DialerOption func(*AteletDialer)

DialerOption customizes an AteletDialer built by NewAteletDialer.

func WithDialCredentials 

func WithDialCredentials(build func(expectedPodUID string) (credentials.TransportCredentials, error)) DialerOption

WithDialCredentials overrides how transport credentials are built for a given atelet pod UID. Tests use it to reach a fake atelet over insecure transport while still exercising the real lookup, dial and connection-cache path.

type Instruments 

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

Instruments holds ateapi's actor-lifecycle and scheduler duration histograms. A nil *Instruments is a valid no-op, so call sites need no guard. Worker-count is registered separately (RegisterWorkerCount): a callback-driven observable, not a synchronous instrument.

func NewInstruments 

func NewInstruments(meter metric.Meter) (*Instruments, error)

NewInstruments builds the two histograms against meter. Assignment buckets are finer than lifecycle's (a cache pick plus a few store writes, not a multi-second restore) but reach 5s so store latency spikes stay measurable.

type RPCService 

type RPCService struct {
	ateapipb.UnimplementedControlServer
	// contains filtered or unexported fields
}

RPCService implements ateapipb.ControlServer and provides the implementation of the RPC service.

Methods on this service should be as light as possible, delegating to the ServiceImpl for business logic and invariants.

func NewRPCService 

func NewRPCService(
	persistence store.Interface,
	workerCache *workercache.Cache,
	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
	csiDriverConfigLister listersv1alpha1.CSIDriverConfigLister,
	storageClassLister storagev1listers.StorageClassLister,
	dialer *AteletDialer,
	instruments *Instruments,
	egressGatewayAddress string,
	volumePlugins map[string]volume.VolumePluginControlPlane,
	objectStore objectstore.Store,
	actorJWTIssuer string,
	actorIDJWTPool localjwtauthority.Pool,
	actorIDCAPool localca.Pool,
) *RPCService

NewRPCService creates an instance of the ControlServer service. This is what implements the outward-facing RPC interface.

instruments may be nil; the record helpers no-op.

objectStore may be nil, which leaves external snapshots in place instead of copying and releasing them. Only tests that never reach those steps pass nil; ate-api always builds one.

actorJWTIssuer is copied verbatim into the iss claim of every actor JWT.

func (*RPCService) CreateActor 

func (s *RPCService) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error)

func (*RPCService) CreateActorEgressPolicy 

func (s *RPCService) CreateActorEgressPolicy(ctx context.Context, req *ateapipb.CreateActorEgressPolicyRequest) (*ateapipb.EgressPolicy, error)

func (*RPCService) CreateActorTemplate 

func (s *RPCService) CreateActorTemplate(ctx context.Context, req *ateapipb.CreateActorTemplateRequest) (*ateapipb.ActorTemplate, error)

func (*RPCService) CreateAtespace 

func (s *RPCService) CreateAtespace(ctx context.Context, req *ateapipb.CreateAtespaceRequest) (*ateapipb.Atespace, error)

func (*RPCService) CreateTag 

func (s *RPCService) CreateTag(ctx context.Context, req *ateapipb.CreateTagRequest) (*ateapipb.Tag, error)

CreateTag tags the external snapshot a suspended Actor holds, giving the tag its own copy of that snapshot so the Actor being suspended again or deleted cannot collect it. The work is a workflow because it spans two transactions around an object copy; see TagActorSnapshot.

func (*RPCService) CreateWorker 

func (s *RPCService) CreateWorker(ctx context.Context, req *ateapipb.CreateWorkerRequest) (*ateapipb.Worker, error)

func (*RPCService) DeleteActor 

func (s *RPCService) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (deleted *ateapipb.Actor, err error)

func (*RPCService) DeleteActorEgressPolicy 

func (s *RPCService) DeleteActorEgressPolicy(ctx context.Context, req *ateapipb.DeleteActorEgressPolicyRequest) (*ateapipb.EgressPolicy, error)

func (*RPCService) DeleteActorTemplate 

func (s *RPCService) DeleteActorTemplate(ctx context.Context, req *ateapipb.DeleteActorTemplateRequest) (*ateapipb.ActorTemplate, error)

func (*RPCService) DeleteAtespace 

func (s *RPCService) DeleteAtespace(ctx context.Context, req *ateapipb.DeleteAtespaceRequest) (*ateapipb.Atespace, error)

func (*RPCService) DeleteTag 

func (s *RPCService) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagRequest) (*ateapipb.Tag, error)

DeleteTag removes the tag and collects the external snapshot it owns.

func (*RPCService) DeleteWorker 

func (s *RPCService) DeleteWorker(ctx context.Context, req *ateapipb.DeleteWorkerRequest) (*ateapipb.Worker, error)

func (*RPCService) DrainWorker 

func (s *RPCService) DrainWorker(ctx context.Context, req *ateapipb.DrainWorkerRequest) (*ateapipb.Worker, error)

func (*RPCService) GetActor 

func (s *RPCService) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (*ateapipb.Actor, error)

func (*RPCService) GetActorEgressPolicy 

func (s *RPCService) GetActorEgressPolicy(ctx context.Context, req *ateapipb.GetActorEgressPolicyRequest) (*ateapipb.EgressPolicy, error)

func (*RPCService) GetActorTemplate 

func (s *RPCService) GetActorTemplate(ctx context.Context, req *ateapipb.GetActorTemplateRequest) (*ateapipb.ActorTemplate, error)

func (*RPCService) GetAtespace 

func (s *RPCService) GetAtespace(ctx context.Context, req *ateapipb.GetAtespaceRequest) (*ateapipb.Atespace, error)

func (*RPCService) GetPlugin 

func (s *RPCService) GetPlugin(ctx context.Context, driverName string) (volume.VolumePluginControlPlane, error)

GetPlugin retrieves a CSI volume plugin by driver name, dynamically discovering it if not present.

func (*RPCService) GetTag 

func (s *RPCService) GetTag(ctx context.Context, req *ateapipb.GetTagRequest) (*ateapipb.Tag, error)

func (*RPCService) GetWorker 

func (s *RPCService) GetWorker(ctx context.Context, req *ateapipb.GetWorkerRequest) (*ateapipb.Worker, error)

func (*RPCService) ListActorTemplates 

func (s *RPCService) ListActorTemplates(ctx context.Context, req *ateapipb.ListActorTemplatesRequest) (*ateapipb.ListActorTemplatesResponse, error)

func (*RPCService) ListActors 

func (s *RPCService) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error)

func (*RPCService) ListAtespaces 

func (s *RPCService) ListAtespaces(ctx context.Context, req *ateapipb.ListAtespacesRequest) (*ateapipb.ListAtespacesResponse, error)

func (*RPCService) ListTags 

func (s *RPCService) ListTags(ctx context.Context, req *ateapipb.ListTagsRequest) (*ateapipb.ListTagsResponse, error)

func (*RPCService) ListWorkerActorAssignments 

func (s *RPCService) ListWorkerActorAssignments(ctx context.Context, req *ateapipb.ListWorkerActorAssignmentsRequest) (*ateapipb.ListWorkerActorAssignmentsResponse, error)

ListWorkerActorAssignments lists the Actors a Worker hosts. The assignments are a subresource rather than a field on Worker, so this is the only way to read them and neither GetWorker nor ListWorkers grows with occupancy.

func (*RPCService) ListWorkers 

func (s *RPCService) ListWorkers(ctx context.Context, req *ateapipb.ListWorkersRequest) (*ateapipb.ListWorkersResponse, error)

func (*RPCService) MintActorCertificate added in v0.2.0 

func (s *RPCService) MintActorCertificate(ctx context.Context, req *ateapipb.MintActorCertificateRequest) (*ateapipb.MintActorCertificateResponse, error)

func (*RPCService) MintActorJWT added in v0.2.0 

func (s *RPCService) MintActorJWT(ctx context.Context, req *ateapipb.MintActorJWTRequest) (*ateapipb.MintActorJWTResponse, error)

func (*RPCService) PauseActor 

func (s *RPCService) PauseActor(ctx context.Context, req *ateapipb.PauseActorRequest) (*ateapipb.PauseActorResponse, error)

func (*RPCService) ResumeActor 

func (s *RPCService) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error)

func (*RPCService) RevertActor added in v0.2.0 

func (s *RPCService) RevertActor(ctx context.Context, req *ateapipb.RevertActorRequest) (*ateapipb.RevertActorResponse, error)

func (*RPCService) SuspendActor 

func (s *RPCService) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error)

func (*RPCService) UpdateActor 

func (s *RPCService) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequest) (*ateapipb.Actor, error)

func (*RPCService) UpdateActorEgressPolicy 

func (s *RPCService) UpdateActorEgressPolicy(ctx context.Context, req *ateapipb.UpdateActorEgressPolicyRequest) (*ateapipb.EgressPolicy, error)

func (*RPCService) UpdateTag 

func (s *RPCService) UpdateTag(ctx context.Context, req *ateapipb.UpdateTagRequest) (*ateapipb.Tag, error)

func (*RPCService) UpdateWorker 

func (s *RPCService) UpdateWorker(ctx context.Context, req *ateapipb.UpdateWorkerRequest) (*ateapipb.Worker, error)

UpdateWorker replaces the stored Worker with the one the request carries. Only labels are the caller's to change; a request that alters an immutable field — including by leaving it unset, which would clear it — is rejected. The service layer enforces that with declarative validation against the stored worker inside the update transaction.

type ServiceImpl 

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

ServiceImpl implements store.Interface and provides the "middleware" layer between the RPC and storage layers. It enforces invariants and validation rules, and may implement additional logic beyond the storage layer.

Methods on this service should hold most of the logic.

func (*ServiceImpl) AcquireLease 

func (s *ServiceImpl) AcquireLease(ctx context.Context, key string) (*store.Lease, error)

Pass-through.

func (*ServiceImpl) BindActorToWorker 

func (s *ServiceImpl) BindActorToWorker(ctx context.Context, workerName string, assignment *ateapipb.ActorAssignment, admit func(*ateapipb.Worker) error) error

The assignment operations are pass-throughs: an assignment is its own record, so binding and releasing are single store calls rather than a read-modify-write of the Worker.

func (*ServiceImpl) CreateActor 

func (s *ServiceImpl) CreateActor(ctx context.Context, inActor *ateapipb.Actor) (*ateapipb.Actor, error)

func (*ServiceImpl) CreateActorTemplate 

func (s *ServiceImpl) CreateActorTemplate(ctx context.Context, inTemplate *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error)

func (*ServiceImpl) CreateAtespace 

func (s *ServiceImpl) CreateAtespace(ctx context.Context, inAtespace *ateapipb.Atespace) (*ateapipb.Atespace, error)

func (*ServiceImpl) CreateEgressPolicy 

func (s *ServiceImpl) CreateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, policy *ateapipb.EgressPolicy) (*ateapipb.EgressPolicy, error)

func (*ServiceImpl) CreateTag 

func (s *ServiceImpl) CreateTag(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error)

func (*ServiceImpl) CreateWorker 

func (s *ServiceImpl) CreateWorker(ctx context.Context, inWorker *ateapipb.Worker) (*ateapipb.Worker, error)

func (*ServiceImpl) DeleteActor 

func (s *ServiceImpl) DeleteActor(ctx context.Context, actorRef resources.ActorRef, precondition store.DeletePreconditions) (*ateapipb.Actor, error)

func (*ServiceImpl) DeleteActorTemplate 

func (s *ServiceImpl) DeleteActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, precondition store.DeletePreconditions) (*ateapipb.ActorTemplate, error)

func (*ServiceImpl) DeleteAtespace 

func (s *ServiceImpl) DeleteAtespace(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.Atespace, error)

func (*ServiceImpl) DeleteEgressPolicy 

func (s *ServiceImpl) DeleteEgressPolicy(ctx context.Context, actorRef resources.ActorRef, precondition store.DeletePreconditions) (*ateapipb.EgressPolicy, error)

func (*ServiceImpl) DeleteTag 

func (s *ServiceImpl) DeleteTag(ctx context.Context, tagRef resources.TagRef, precondition store.DeletePreconditions) (*ateapipb.Tag, error)

func (*ServiceImpl) DeleteWorker 

func (s *ServiceImpl) DeleteWorker(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.Worker, error)

func (*ServiceImpl) FindWorkerHostingActor 

func (s *ServiceImpl) FindWorkerHostingActor(ctx context.Context, actorUID string) (string, error)

func (*ServiceImpl) GetActor 

func (s *ServiceImpl) GetActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error)

func (*ServiceImpl) GetActorTemplate 

func (s *ServiceImpl) GetActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)

func (*ServiceImpl) GetAtespace 

func (s *ServiceImpl) GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error)

func (*ServiceImpl) GetEgressPolicy 

func (s *ServiceImpl) GetEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)

func (*ServiceImpl) GetTag 

func (s *ServiceImpl) GetTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)

func (*ServiceImpl) GetWorker 

func (s *ServiceImpl) GetWorker(ctx context.Context, name string) (*ateapipb.Worker, error)

GetWorker returns the Worker with the Actors it hosts, read separately since the assignments are their own records. The one read that pays O(assignments). A failed read is an error, not a Worker reported as hosting nothing: the caller cannot tell those apart.

func (*ServiceImpl) GetWorkerAssignment 

func (s *ServiceImpl) GetWorkerAssignment(ctx context.Context, workerName, actorUID string) (*ateapipb.ActorAssignment, error)

func (*ServiceImpl) ListActorTemplates 

func (s *ServiceImpl) ListActorTemplates(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorTemplate], error)

func (*ServiceImpl) ListActors 

func (s *ServiceImpl) ListActors(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Actor], error)

func (*ServiceImpl) ListAtespaces 

func (s *ServiceImpl) ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error)

func (*ServiceImpl) ListTags 

func (s *ServiceImpl) ListTags(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Tag], error)

func (*ServiceImpl) ListWorkerAssignments 

func (s *ServiceImpl) ListWorkerAssignments(ctx context.Context, workerName string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorAssignment], error)

func (*ServiceImpl) ListWorkers 

func (s *ServiceImpl) ListWorkers(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Worker], error)

func (*ServiceImpl) ReleaseActorFromWorker 

func (s *ServiceImpl) ReleaseActorFromWorker(ctx context.Context, workerName string, actorUID string) (*ateapipb.Worker, error)

func (*ServiceImpl) UpdateActor 

func (s *ServiceImpl) UpdateActor(ctx context.Context, actorRef resources.ActorRef, precondition store.Precondition, mutate func(*ateapipb.Actor) error) (*ateapipb.Actor, error)

func (*ServiceImpl) UpdateActorTemplate 

func (s *ServiceImpl) UpdateActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, precondition store.Precondition, mutate func(dbTemplate *ateapipb.ActorTemplate) error) (*ateapipb.ActorTemplate, error)

func (*ServiceImpl) UpdateEgressPolicy 

func (s *ServiceImpl) UpdateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, precondition store.Precondition, mutate func(*ateapipb.EgressPolicy) error) (*ateapipb.EgressPolicy, error)

func (*ServiceImpl) UpdateTag 

func (s *ServiceImpl) UpdateTag(ctx context.Context, tagRef resources.TagRef, precondition store.Precondition, mutate func(toUpdate *ateapipb.Tag) error) (*ateapipb.Tag, error)

func (*ServiceImpl) UpdateWorker 

func (s *ServiceImpl) UpdateWorker(ctx context.Context, name string, precondition store.Precondition, mutate func(toUpdate *ateapipb.Worker) error) (*ateapipb.Worker, error)

func (*ServiceImpl) WatchWorkers 

func (s *ServiceImpl) WatchWorkers(ctx context.Context) (*store.WorkerWatch, error)

type VolumePluginRegistry 

type VolumePluginRegistry interface {
	GetPlugin(ctx context.Context, name string) (volume.VolumePluginControlPlane, error)
}

VolumePluginRegistry defines the interface for dynamic CSI plugin resolution.

type WorkerWorkflow 

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

WorkerWorkflow handles the multi-step operations on a Worker.

Its steps reach across the Actor↔Worker binding, which an ActorWorkflow step does from the other side: releasing the Actor bound to a Worker stays in-process because there is no bind/release RPC.

func NewWorkerWorkflow 

func NewWorkerWorkflow(store workerWorkflowStore) *WorkerWorkflow

NewWorkerWorkflow creates a new WorkerWorkflow.

func (*WorkerWorkflow) DeleteWorker 

func (w *WorkerWorkflow) DeleteWorker(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.Worker, error)

DeleteWorker executes the workflow to deregister a Worker. The caller reaches here because the Worker's pod is gone, so the Actor bound to it — if any — has lost its sandbox and is released before the record is removed.

Re-drivable in the sense DeleteActor is: a failed attempt leaves the Worker record in place, and a retry fast-forwards past whatever the previous attempt already did. An absent Worker is NOT_FOUND rather than success; idempotency belongs to the caller, which knows whether that is the state it wanted.

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.