distributedtx

Documentation

Index

• Constants
• Variables
• func ResourceLockRel(input *WriteObjInput, workflowID string) *v1.RelationshipUpdate
• func WorkflowForLockMode(lockMode string) (any, error)
• type ActivityHandler
  • func (h *ActivityHandler) CheckKubeResource(ctx context.Context, req *KubeReqInput) (bool, error)
  • func (h *ActivityHandler) WriteToKube(ctx context.Context, req *KubeReqInput) (*KubeResp, error)
  • func (h *ActivityHandler) WriteToSpiceDB(ctx context.Context, input *v1.WriteRelationshipsRequest) (*v1.WriteRelationshipsResponse, error)
• type KubeReqInput
• type KubeResp
  • func KubeConflict(err error, input *WriteObjInput) *KubeResp
  • func OptimisticWriteToSpiceDBAndKube(ctx workflow.Context, input *WriteObjInput) (*KubeResp, error)
  • func PessimisticWriteToSpiceDBAndKube(ctx workflow.Context, input *WriteObjInput) (*KubeResp, error)
• type RollbackRelationships
  • func NewRollbackRelationships(rels ...*v1.RelationshipUpdate) *RollbackRelationships
  • func (r *RollbackRelationships) Cleanup(ctx workflow.Context)
  • func (r *RollbackRelationships) WithRels(relationships ...*v1.RelationshipUpdate) *RollbackRelationships
• type Worker
  • func SetupWithBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, ...) (*client.Client, *Worker, error)
  • func SetupWithMemoryBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, ...) (*client.Client, *Worker, error)
  • func SetupWithSQLiteBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, ...) (*client.Client, *Worker, error)
  • func (w *Worker) Shutdown(_ context.Context) error
  • func (w *Worker) Start(ctx context.Context) error
• type WriteObjInput

Constants

const (
	MaxKubeAttempts                          = 5
	StrategyOptimisticWriteToSpiceDBAndKube  = "Optimistic"
	StrategyPessimisticWriteToSpiceDBAndKube = "Pessimistic"
	DefaultWorkflowTimeout                   = time.Second * 30
)

Variables

var KubeBackoff = wait.Backoff{
	Duration: 100 * time.Millisecond,
	Factor:   2,
	Jitter:   0.1,
	Steps:    MaxKubeAttempts,
}

Functions

func ResourceLockRel

func ResourceLockRel(input *WriteObjInput, workflowID string) *v1.RelationshipUpdate

ResourceLockRel generates a relationship representing a worfklow's lock over a specific resource in kube.

func WorkflowForLockMode

func WorkflowForLockMode(lockMode string) (any, error)

Types

type ActivityHandler

type ActivityHandler struct {
	PermissionClient v1.PermissionsServiceClient
	KubeClient       rest.Interface
}

func (*ActivityHandler) CheckKubeResource

func (h *ActivityHandler) CheckKubeResource(ctx context.Context, req *KubeReqInput) (bool, error)

func (*ActivityHandler) WriteToKube

func (h *ActivityHandler) WriteToKube(ctx context.Context, req *KubeReqInput) (*KubeResp, error)

WriteToKube performs a Kube API Server POST, specified in a KubeReqInput

func (*ActivityHandler) WriteToSpiceDB

func (h *ActivityHandler) WriteToSpiceDB(ctx context.Context, input *v1.WriteRelationshipsRequest) (*v1.WriteRelationshipsResponse, error)

WriteToSpiceDB writes relationships to spicedb and returns any errors.

type KubeReqInput

type KubeReqInput struct {
	RequestInfo *request.RequestInfo
	Header      http.Header
	ObjectMeta  *metav1.ObjectMeta
	Body        []byte
}

type KubeResp

type KubeResp struct {
	Body        []byte
	ContentType string
	StatusCode  int
	Err         k8serrors.StatusError
}

func KubeConflict

func KubeConflict(err error, input *WriteObjInput) *KubeResp

KubeConflict wraps an error and turns it into a standard kube conflict response.

func OptimisticWriteToSpiceDBAndKube

func OptimisticWriteToSpiceDBAndKube(ctx workflow.Context, input *WriteObjInput) (*KubeResp, error)

OptimisticWriteToSpiceDBAndKube ensures that a write exists in both SpiceDB and kube, or neither. It attempts to perform the writes and rolls back if errors are encountered, leaving the user to retry on write conflicts.

func PessimisticWriteToSpiceDBAndKube

func PessimisticWriteToSpiceDBAndKube(ctx workflow.Context, input *WriteObjInput) (*KubeResp, error)

PessimisticWriteToSpiceDBAndKube ensures that a write exists in both SpiceDB and kube, or neither, using locks. It prevents multiple users from writing the same object/fields at the same time

type RollbackRelationships

type RollbackRelationships []*v1.RelationshipUpdate

func NewRollbackRelationships

func NewRollbackRelationships(rels ...*v1.RelationshipUpdate) *RollbackRelationships

func (*RollbackRelationships) Cleanup

func (r *RollbackRelationships) Cleanup(ctx workflow.Context)

func (*RollbackRelationships) WithRels

func (r *RollbackRelationships) WithRels(relationships ...*v1.RelationshipUpdate) *RollbackRelationships

type Worker

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

func SetupWithBackend

func SetupWithBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, kubeClient rest.Interface, backend backend.Backend) (*client.Client, *Worker, error)

func SetupWithMemoryBackend

func SetupWithMemoryBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, kubeClient rest.Interface) (*client.Client, *Worker, error)

func SetupWithSQLiteBackend

func SetupWithSQLiteBackend(ctx context.Context, permissionClient v1.PermissionsServiceClient, kubeClient rest.Interface, sqlitePath string) (*client.Client, *Worker, error)

func (*Worker) Shutdown

func (w *Worker) Shutdown(_ context.Context) error

func (*Worker) Start

func (w *Worker) Start(ctx context.Context) error

type WriteObjInput

type WriteObjInput struct {
	RequestInfo   *request.RequestInfo
	Header        http.Header
	UserInfo      *user.DefaultInfo
	ObjectMeta    *metav1.ObjectMeta
	Rels          []*v1.Relationship
	Preconditions []*v1.Precondition
	Body          []byte
}