eventutil

Documentation

Index

• Constants
• func ApiJobFromLogSubmitJob(ownerId string, groups []string, queueName string, jobSetName string, ...) (*api.Job, error)
• func ApiJobsFromLogSubmitJobs(userId string, groups []string, queueName string, jobSetName string, ...) ([]*api.Job, error)
• func CompactEventSequences(sequences []*armadaevents.EventSequence) []*armadaevents.EventSequence
• func EventSequenceFromApiEvent(msg *api.EventMessage) (sequence *armadaevents.EventSequence, err error)
• func EventSequencesFromApiEvents(msgs []*api.EventMessage) ([]*armadaevents.EventSequence, error)
• func K8sObjectMetaFromLogObjectMeta(meta *armadaevents.ObjectMeta) *metav1.ObjectMeta
• func K8sServicesIngressesFromApiJob(job *api.Job, ingressConfig *configuration.IngressConfiguration) ([]*v1.Service, []*networking.Ingress, error)
• func LegacyJobRunId() *armadaevents.Uuid
• func LimitSequenceByteSize(sequence *armadaevents.EventSequence, sizeInBytes uint, strict bool) ([]*armadaevents.EventSequence, error)
• func LimitSequencesByteSize(sequences []*armadaevents.EventSequence, sizeInBytes uint, strict bool) ([]*armadaevents.EventSequence, error)
• func LogObjectMetaFromK8sObjectMeta(meta *metav1.ObjectMeta) *armadaevents.ObjectMeta
• func LogSubmitJobFromApiJob(job *api.Job) (*armadaevents.SubmitJob, error)
• func LogSubmitObjectsFromApiJob(job *api.Job) (*armadaevents.KubernetesMainObject, []*armadaevents.KubernetesObject, error)
• func LogSubmitPriorityFromApiPriority(priority float64) uint32
• func PopulateK8sServicesIngresses(job *api.Job, ingressConfig *configuration.IngressConfiguration) error
• func ShortSequenceString(sequence *armadaevents.EventSequence) string
• func UnmarshalEventSequence(ctx context.Context, payload []byte) (*armadaevents.EventSequence, error)
• type EventFilter
  • func NewEventFilter(in chan *EventSequenceWithMessageIds, ...) *EventFilter
  • func (srv *EventFilter) Run(ctx context.Context) error
• type EventSequenceWithMessageIds
• type SequenceCompacter
  • func NewSequenceCompacter(in chan *EventSequenceWithMessageIds) *SequenceCompacter
  • func (srv *SequenceCompacter) Run(ctx context.Context) error
• type SequenceFromMessage
  • func NewSequenceFromMessage(in chan pulsar.Message) *SequenceFromMessage
  • func (srv *SequenceFromMessage) Run(ctx context.Context) error

Constants

const LEGACY_RUN_ID = "00000000-0000-0000-0000-000000000000"

LEGACY_RUN_ID is used for messages for which we can't use the kubernetesId.

Functions

func ApiJobFromLogSubmitJob

func ApiJobFromLogSubmitJob(ownerId string, groups []string, queueName string, jobSetName string, time time.Time, e *armadaevents.SubmitJob) (*api.Job, error)

ApiJobFromLogSubmitJob converts a SubmitJob log message into an api.Job struct, which is used by Armada internally.

func ApiJobsFromLogSubmitJobs

func ApiJobsFromLogSubmitJobs(
	userId string,
	groups []string,
	queueName string,
	jobSetName string,
	time time.Time,
	es []*armadaevents.SubmitJob,
) ([]*api.Job, error)

ApiJobsFromLogSubmitJobs converts a slice of log jobs to API jobs.

func CompactEventSequences

func CompactEventSequences(sequences []*armadaevents.EventSequence) []*armadaevents.EventSequence

CompactEventSequences converts a []*armadaevents.EventSequence into a []*armadaevents.EventSequence of minimal length. In particular, it moves events with equal (queue, jobSetName, userId, groups) into a single sequence when doing so is possible without changing the order of events within job sets.

For example, three sequences [A, B, C], [D, E], [F, G] could result in the following two sequences [A, B, C, F, G], [D, E], if sequence 1 and 3 share the same (queue, jobSetName, userId, groups) and if the sequence [D, E] is for a different job set.

func EventSequenceFromApiEvent

func EventSequenceFromApiEvent(msg *api.EventMessage) (sequence *armadaevents.EventSequence, err error)

EventSequenceFromApiEvent converts an api.EventMessage into the corresponding Pulsar event and returns an EventSequence containing this single event. We map API events to sequences one-to-one because each API event may contain different (queue, jobSet, userId), which must be common to all events in a sequence.

func EventSequencesFromApiEvents

func EventSequencesFromApiEvents(msgs []*api.EventMessage) ([]*armadaevents.EventSequence, error)

func K8sObjectMetaFromLogObjectMeta

func K8sObjectMetaFromLogObjectMeta(meta *armadaevents.ObjectMeta) *metav1.ObjectMeta

func K8sServicesIngressesFromApiJob

func K8sServicesIngressesFromApiJob(job *api.Job, ingressConfig *configuration.IngressConfiguration) ([]*v1.Service, []*networking.Ingress, error)

K8sServicesIngressesFromApiJob converts job.Services and job.Ingress to k8s services and ingresses.

func LegacyJobRunId

func LegacyJobRunId() *armadaevents.Uuid

func LimitSequenceByteSize

func LimitSequenceByteSize(sequence *armadaevents.EventSequence, sizeInBytes uint, strict bool) ([]*armadaevents.EventSequence, error)

LimitSequenceByteSize returns a slice of sequences produced by breaking up sequence.Events into separate sequences, each of which is at most MAX_SEQUENCE_SIZE_IN_BYTES bytes in size.

func LimitSequencesByteSize

func LimitSequencesByteSize(sequences []*armadaevents.EventSequence, sizeInBytes uint, strict bool) ([]*armadaevents.EventSequence, error)

LimitSequencesByteSize calls LimitSequenceByteSize for each of the provided sequences and returns all resulting sequences.

func LogObjectMetaFromK8sObjectMeta

func LogObjectMetaFromK8sObjectMeta(meta *metav1.ObjectMeta) *armadaevents.ObjectMeta

func LogSubmitJobFromApiJob

func LogSubmitJobFromApiJob(job *api.Job) (*armadaevents.SubmitJob, error)

LogSubmitJobFromApiJob converts an API job to a log job. Note that PopulateK8sServicesIngresses must be called first if job.Services and job.Ingress is to be included in the resulting log job, since the log job can only include k8s objects (i.e., not the API-specific job.Services or job.Ingress).

func LogSubmitObjectsFromApiJob

func LogSubmitObjectsFromApiJob(job *api.Job) (*armadaevents.KubernetesMainObject, []*armadaevents.KubernetesObject, error)

LogSubmitObjectsFromApiJob extracts all objects from an API job for inclusion in a log job.

To extract services and ingresses, PopulateK8sServicesIngresses must be called on the job first to convert API-specific job objects to proper K8s objects.

func LogSubmitPriorityFromApiPriority

func LogSubmitPriorityFromApiPriority(priority float64) uint32

LogSubmitPriorityFromApiPriority returns the uint32 representation of the priority included with a submitted job, or an error if the conversion fails.

func PopulateK8sServicesIngresses

func PopulateK8sServicesIngresses(job *api.Job, ingressConfig *configuration.IngressConfiguration) error

PopulateK8sServicesIngresses converts the API-specific service and ingress object into K8s objects and stores those in the job object.

func ShortSequenceString

func ShortSequenceString(sequence *armadaevents.EventSequence) string

ShortSequenceString returns a short string representation of an events sequence. To be used for logging, for example.

func UnmarshalEventSequence

func UnmarshalEventSequence(ctx context.Context, payload []byte) (*armadaevents.EventSequence, error)

UnmarshalEventSequence returns an EventSequence object contained in a byte buffer after validating that the resulting EventSequence is valid.

Types

type EventFilter

type EventFilter struct {
	In  chan *EventSequenceWithMessageIds
	Out chan *EventSequenceWithMessageIds
	// contains filtered or unexported fields
}

EventFilter calls filter once for each event, and events for which filter returns false are discarded.

func NewEventFilter

func NewEventFilter(in chan *EventSequenceWithMessageIds, filter func(*armadaevents.EventSequence_Event) bool) *EventFilter

func (*EventFilter) Run

func (srv *EventFilter) Run(ctx context.Context) error

type EventSequenceWithMessageIds

type EventSequenceWithMessageIds struct {
	Sequence   *armadaevents.EventSequence
	MessageIds []pulsar.MessageID
}

EventSequenceWithMessageIds bundles an event sequence with all the ids of all Pulsar messages that were consumed to produce it.

type SequenceCompacter

type SequenceCompacter struct {
	In  chan *EventSequenceWithMessageIds
	Out chan *EventSequenceWithMessageIds
	// Buffer messages for at most this long before forwarding on the outgoing channel.
	Interval time.Duration
	// Max number of events to buffer.
	MaxEvents int
	// contains filtered or unexported fields
}

SequenceCompacter reads sequences and produces compacted sequences. Compacted sequences are created by combining events in sequences with the

func NewSequenceCompacter

func NewSequenceCompacter(in chan *EventSequenceWithMessageIds) *SequenceCompacter

func (*SequenceCompacter) Run

func (srv *SequenceCompacter) Run(ctx context.Context) error

type SequenceFromMessage

type SequenceFromMessage struct {
	In  chan pulsar.Message
	Out chan *EventSequenceWithMessageIds
}

PulsarToChannel is a service for receiving messages from Pulsar and forwarding those on C.

func NewSequenceFromMessage

func NewSequenceFromMessage(in chan pulsar.Message) *SequenceFromMessage

func (*SequenceFromMessage) Run

func (srv *SequenceFromMessage) Run(ctx context.Context) error