Documentation
¶
Index ¶
- func CalculateMaxLookback(counts []*TimestampedCounts, startIndex, endIndex int) int64
- func CalculatePending(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) int64
- func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) float64
- type MonoVtxRatable
- type Option
- type PodInfo
- type PodMetricsCount
- type PodTracker
- type PodTrackerOption
- type Rater
- type TimestampedCounts
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CalculateMaxLookback ¶ added in v1.5.0
func CalculateMaxLookback(counts []*TimestampedCounts, startIndex, endIndex int) int64
CalculateMaxLookback computes the maximum duration (in seconds) for which the count of messages processed by any pod remained unchanged within a specified range of indices in a queue of TimestampedCounts. It does this by analyzing each data point between the startIndex and endIndex, checking the count changes for each pod, and noting the durations during which these counts stay consistent. The metric is updated when data is read by the pod This would encapsulate the lookback for two scenarios 1. Slow processing vertex 2. Slow data source - data arrives after long intervals
func CalculatePending ¶ added in v1.5.0
func CalculatePending(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) int64
CalculatePending calculates the pending of a MonoVertex for a given lookback period.
func CalculateRate ¶
func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) float64
CalculateRate calculates the rate of a MonoVertex for a given lookback period.
Types ¶
type MonoVtxRatable ¶
type MonoVtxRatable interface {
Start(ctx context.Context) error
GetRates() map[string]*wrapperspb.DoubleValue
GetPending() map[string]*wrapperspb.Int64Value
}
MonoVtxRatable is the interface for the Rater struct.
type PodInfo ¶ added in v1.7.0
type PodInfo struct {
// contains filtered or unexported fields
}
PodInfo represents the information of a pod that is used for tracking the processing rate
type PodMetricsCount ¶ added in v1.5.0
type PodMetricsCount struct {
// contains filtered or unexported fields
}
PodMetricsCount is a struct to maintain count of messages read by a pod of MonoVertex
func (*PodMetricsCount) Name ¶ added in v1.5.0
func (p *PodMetricsCount) Name() string
Name returns the pod name
func (*PodMetricsCount) ReadCount ¶ added in v1.5.0
func (p *PodMetricsCount) ReadCount() float64
ReadCount returns the value of the messages read by the Pod
type PodTracker ¶
type PodTracker struct {
// contains filtered or unexported fields
}
PodTracker maintains a set of active pods for a MonoVertex It periodically sends http requests to pods to check if they are still active
func NewPodTracker ¶
func NewPodTracker(ctx context.Context, mv *v1alpha1.MonoVertex, opts ...PodTrackerOption) *PodTracker
func (*PodTracker) GetActivePodsCount ¶
func (pt *PodTracker) GetActivePodsCount() int
GetActivePodsCount returns the number of active pods.
func (*PodTracker) GetPodInfo ¶
func (pt *PodTracker) GetPodInfo(key string) (*PodInfo, error)
func (*PodTracker) IsActive ¶
func (pt *PodTracker) IsActive(podKey string) bool
IsActive returns true if the pod is active, false otherwise.
func (*PodTracker) LeastRecentlyUsed ¶
func (pt *PodTracker) LeastRecentlyUsed() string
LeastRecentlyUsed returns the least recently used pod from the active pod list. if there are no active pods, it returns an empty string.
type PodTrackerOption ¶
type PodTrackerOption func(*PodTracker)
func WithRefreshInterval ¶
func WithRefreshInterval(d time.Duration) PodTrackerOption
WithRefreshInterval sets how often to refresh the rate metrics.
type Rater ¶
type Rater struct {
// contains filtered or unexported fields
}
Rater is a struct that maintains information about the processing rate of the MonoVertex. It monitors the number of processed messages for each pod in a MonoVertex and calculates the rate.
func (*Rater) GetPending ¶ added in v1.5.0
func (r *Rater) GetPending() map[string]*wrapperspb.Int64Value
GetPending returns the pending count for the mono vertex
func (*Rater) GetRates ¶
func (r *Rater) GetRates() map[string]*wrapperspb.DoubleValue
GetRates returns the rate metrics for the MonoVertex. It calculates the rate metrics for the given lookback seconds.
type TimestampedCounts ¶
type TimestampedCounts struct {
// contains filtered or unexported fields
}
TimestampedCounts track the total count of processed messages for a list of pods at a given timestamp
func NewTimestampedCounts ¶
func NewTimestampedCounts(t int64) *TimestampedCounts
func (*TimestampedCounts) PodCountSnapshot ¶
func (tc *TimestampedCounts) PodCountSnapshot() map[string]float64
PodCountSnapshot returns a copy of podReadCounts it's used to ensure the returned map is not modified by other goroutines
func (*TimestampedCounts) PodTimestamp ¶ added in v1.5.0
func (tc *TimestampedCounts) PodTimestamp() int64
func (*TimestampedCounts) String ¶
func (tc *TimestampedCounts) String() string
String returns a string representation of the TimestampedCounts it's used for debugging purpose
func (*TimestampedCounts) Update ¶
func (tc *TimestampedCounts) Update(podReadCount *PodMetricsCount)
Update updates the count of processed messages for a pod
Source Files