yusheng-guo/kube-metrics-adapter
1
0
Fork 0
mirror of https://github.com/zalando-incubator/kube-metrics-adapter.git synced 2024-12-22 11:06:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move generic functions to controller package

Browse Source 
Move defaultTimeZone to a flag.

Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de>
This commit is contained in:
Mikkel Oscar Lyderik Larsen 2023-04-06 17:45:15 +02:00
parent b631a4fe08
commit e2a922f110
4 changed files with 184 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
pkg/collector/scaling_schedule_collector.go
View File

@ -7,32 +7,13 @@ import (
"time"
v1 "github.com/zalando-incubator/kube-metrics-adapter/pkg/apis/zalando.org/v1"
scheduledscaling "github.com/zalando-incubator/kube-metrics-adapter/pkg/controller/scheduledscaling"
autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/metrics/pkg/apis/custom_metrics"
)
const (
// The format used by v1.SchedulePeriod.StartTime. 15:04 are
// the defined reference time in time.Format.
hourColonMinuteLayout = "15:04"
// The default timezone used in v1.SchedulePeriod if none is
// defined.
// TODO(jonathanbeber): it should be configurable.
defaultTimeZone = "Europe/Berlin"
)
var days = map[v1.ScheduleDay]time.Weekday{
v1.SundaySchedule: time.Sunday,
v1.MondaySchedule: time.Monday,
v1.TuesdaySchedule: time.Tuesday,
v1.WednesdaySchedule: time.Wednesday,
v1.ThursdaySchedule: time.Thursday,
v1.FridaySchedule: time.Friday,
v1.SaturdaySchedule: time.Saturday,
}
var (
// ErrScalingScheduleNotFound is returned when a item referenced in
// the HPA config is not in the ScalingScheduleCollectorPlugin.store.
@ -49,15 +30,6 @@ var (
// be an ClusterScalingSchedule but the type assertion failed. When
// returned the type assertion to ScalingSchedule failed too.
ErrNotClusterScalingScheduleFound = errors.New("error converting returned object to ClusterScalingSchedule")
// ErrInvalidScheduleDate is returned when the v1.ScheduleDate is
// not a valid RFC3339 date. It shouldn't happen since the
// validation is done by the CRD.
ErrInvalidScheduleDate = errors.New("could not parse the specified schedule date, format is not RFC3339")
// ErrInvalidScheduleStartTime is returned when the
// v1.SchedulePeriod.StartTime is not in the format specified by
// hourColonMinuteLayout. It shouldn't happen since the validation
// is done by the CRD.
ErrInvalidScheduleStartTime = errors.New("could not parse the specified schedule period start time, format is not HH:MM")
)
// Now is the function that returns a time.Time object representing the
@ -82,6 +54,7 @@ type ScalingScheduleCollectorPlugin struct {
store Store
now Now
defaultScalingWindow time.Duration
defaultTimeZone string
rampSteps int
}
@ -91,25 +64,28 @@ type ClusterScalingScheduleCollectorPlugin struct {
store Store
now Now
defaultScalingWindow time.Duration
defaultTimeZone string
rampSteps int
}
// NewScalingScheduleCollectorPlugin initializes a new ScalingScheduleCollectorPlugin.
func NewScalingScheduleCollectorPlugin(store Store, now Now, defaultScalingWindow time.Duration, rampSteps int) (*ScalingScheduleCollectorPlugin, error) {
func NewScalingScheduleCollectorPlugin(store Store, now Now, defaultScalingWindow time.Duration, defaultTimeZone string, rampSteps int) (*ScalingScheduleCollectorPlugin, error) {
return &ScalingScheduleCollectorPlugin{
store: store,
now: now,
defaultScalingWindow: defaultScalingWindow,
defaultTimeZone: defaultTimeZone,
rampSteps: rampSteps,
}, nil
}
// NewClusterScalingScheduleCollectorPlugin initializes a new ClusterScalingScheduleCollectorPlugin.
func NewClusterScalingScheduleCollectorPlugin(store Store, now Now, defaultScalingWindow time.Duration, rampSteps int) (*ClusterScalingScheduleCollectorPlugin, error) {
func NewClusterScalingScheduleCollectorPlugin(store Store, now Now, defaultScalingWindow time.Duration, defaultTimeZone string, rampSteps int) (*ClusterScalingScheduleCollectorPlugin, error) {
return &ClusterScalingScheduleCollectorPlugin{
store: store,
now: now,
defaultScalingWindow: defaultScalingWindow,
defaultTimeZone: defaultTimeZone,
rampSteps: rampSteps,
}, nil
}
@ -118,14 +94,14 @@ func NewClusterScalingScheduleCollectorPlugin(store Store, now Now, defaultScali
// specified HPA. It's the only required method to implement the
// collector.CollectorPlugin interface.
func (c *ScalingScheduleCollectorPlugin) NewCollector(hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
return NewScalingScheduleCollector(c.store, c.defaultScalingWindow, c.rampSteps, c.now, hpa, config, interval)
return NewScalingScheduleCollector(c.store, c.defaultScalingWindow, c.defaultTimeZone, c.rampSteps, c.now, hpa, config, interval)
}
// NewCollector initializes a new cluster wide scaling schedule
// collector from the specified HPA. It's the only required method to
// implement the collector.CollectorPlugin interface.
func (c *ClusterScalingScheduleCollectorPlugin) NewCollector(hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
return NewClusterScalingScheduleCollector(c.store, c.defaultScalingWindow, c.rampSteps, c.now, hpa, config, interval)
return NewClusterScalingScheduleCollector(c.store, c.defaultScalingWindow, c.defaultTimeZone, c.rampSteps, c.now, hpa, config, interval)
}
// ScalingScheduleCollector is a metrics collector for time based
@ -152,11 +128,12 @@ type scalingScheduleCollector struct {
interval time.Duration
config MetricConfig
defaultScalingWindow time.Duration
defaultTimeZone string
rampSteps int
}
// NewScalingScheduleCollector initializes a new ScalingScheduleCollector.
func NewScalingScheduleCollector(store Store, defaultScalingWindow time.Duration, rampSteps int, now Now, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (*ScalingScheduleCollector, error) {
func NewScalingScheduleCollector(store Store, defaultScalingWindow time.Duration, defaultTimeZone string, rampSteps int, now Now, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (*ScalingScheduleCollector, error) {
return &ScalingScheduleCollector{
scalingScheduleCollector{
store: store,
@ -167,13 +144,14 @@ func NewScalingScheduleCollector(store Store, defaultScalingWindow time.Duration
interval: interval,
config: *config,
defaultScalingWindow: defaultScalingWindow,
defaultTimeZone: defaultTimeZone,
rampSteps: rampSteps,
},
}, nil
}
// NewClusterScalingScheduleCollector initializes a new ScalingScheduleCollector.
func NewClusterScalingScheduleCollector(store Store, defaultScalingWindow time.Duration, rampSteps int, now Now, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (*ClusterScalingScheduleCollector, error) {
func NewClusterScalingScheduleCollector(store Store, defaultScalingWindow time.Duration, defaultTimeZone string, rampSteps int, now Now, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (*ClusterScalingScheduleCollector, error) {
return &ClusterScalingScheduleCollector{
scalingScheduleCollector{
store: store,
@ -184,6 +162,7 @@ func NewClusterScalingScheduleCollector(store Store, defaultScalingWindow time.D
interval: interval,
config: *config,
defaultScalingWindow: defaultScalingWindow,
defaultTimeZone: defaultTimeZone,
rampSteps: rampSteps,
},
}, nil
@ -203,7 +182,7 @@ func (c *ScalingScheduleCollector) GetMetrics() ([]CollectedMetric, error) {
if !ok {
return nil, ErrNotScalingScheduleFound
}
return calculateMetrics(scalingSchedule.Spec, c.defaultScalingWindow, c.rampSteps, c.now(), c.objectReference, c.metric)
return calculateMetrics(scalingSchedule.Spec, c.defaultScalingWindow, c.defaultTimeZone, c.rampSteps, c.now(), c.objectReference, c.metric)
}
// GetMetrics is the main implementation for collector.Collector interface
@ -236,7 +215,7 @@ func (c *ClusterScalingScheduleCollector) GetMetrics() ([]CollectedMetric, error
clusterScalingSchedule = v1.ClusterScalingSchedule(*scalingSchedule)
}
return calculateMetrics(clusterScalingSchedule.Spec, c.defaultScalingWindow, c.rampSteps, c.now(), c.objectReference, c.metric)
return calculateMetrics(clusterScalingSchedule.Spec, c.defaultScalingWindow, c.defaultTimeZone, c.rampSteps, c.now(), c.objectReference, c.metric)
}
// Interval returns the interval at which the collector should run.
@ -249,7 +228,7 @@ func (c *ClusterScalingScheduleCollector) Interval() time.Duration {
return c.interval
}
func calculateMetrics(spec v1.ScalingScheduleSpec, defaultScalingWindow time.Duration, rampSteps int, now time.Time, objectReference custom_metrics.ObjectReference, metric autoscalingv2.MetricIdentifier) ([]CollectedMetric, error) {
func calculateMetrics(spec v1.ScalingScheduleSpec, defaultScalingWindow time.Duration, defaultTimeZone string, rampSteps int, now time.Time, objectReference custom_metrics.ObjectReference, metric autoscalingv2.MetricIdentifier) ([]CollectedMetric, error) {
scalingWindowDuration := defaultScalingWindow
if spec.ScalingWindowDurationMinutes != nil {
scalingWindowDuration = time.Duration(*spec.ScalingWindowDurationMinutes) * time.Minute
@ -259,89 +238,12 @@ func calculateMetrics(spec v1.ScalingScheduleSpec, defaultScalingWindow time.Dur
}
value := int64(0)
var scheduledEndTime time.Time
for _, schedule := range spec.Schedules {
switch schedule.Type {
case v1.RepeatingSchedule:
location, err := time.LoadLocation(schedule.Period.Timezone)
if schedule.Period.Timezone == "" || err != nil {
location, err = time.LoadLocation(defaultTimeZone)
if err != nil {
return nil, fmt.Errorf("unexpected error loading default location: %s", err.Error())
}
}
nowInLocation := now.In(location)
weekday := nowInLocation.Weekday()
for _, day := range schedule.Period.Days {
if days[day] == weekday {
parsedStartTime, err := time.Parse(hourColonMinuteLayout, schedule.Period.StartTime)
if err != nil {
return nil, ErrInvalidScheduleStartTime
}
scheduledTime := time.Date(
// v1.SchedulePeriod.StartTime can't define the
// year, month or day, so we compute it as the
// current date in the configured location.
nowInLocation.Year(),
nowInLocation.Month(),
nowInLocation.Day(),
// Hours and minute are configured in the
// v1.SchedulePeriod.StartTime.
parsedStartTime.Hour(),
parsedStartTime.Minute(),
parsedStartTime.Second(),
parsedStartTime.Nanosecond(),
location,
)
// If no end time was provided, set it to equal the start time
if schedule.Period.EndTime == "" {
scheduledEndTime = scheduledTime
} else {
parsedEndTime, err := time.Parse(hourColonMinuteLayout, schedule.Period.EndTime)
if err != nil {
return nil, ErrInvalidScheduleDate
}
scheduledEndTime = time.Date(
// v1.SchedulePeriod.StartTime can't define the
// year, month or day, so we compute it as the
// current date in the configured location.
nowInLocation.Year(),
nowInLocation.Month(),
nowInLocation.Day(),
// Hours and minute are configured in the
// v1.SchedulePeriod.StartTime.
parsedEndTime.Hour(),
parsedEndTime.Minute(),
parsedEndTime.Second(),
parsedEndTime.Nanosecond(),
location,
)
}
value = maxInt64(value, valueForEntry(now, scheduledTime, schedule.Duration(), scheduledEndTime, scalingWindowDuration, rampSteps, schedule.Value))
break
}
}
case v1.OneTimeSchedule:
scheduledTime, err := time.Parse(time.RFC3339, string(*schedule.Date))
if err != nil {
return nil, ErrInvalidScheduleDate
}
// If no end time was provided, set it to equal the start time
if schedule.EndDate == nil || string(*schedule.EndDate) == "" {
scheduledEndTime = scheduledTime
} else {
scheduledEndTime, err = time.Parse(time.RFC3339, string(*schedule.EndDate))
if err != nil {
return nil, ErrInvalidScheduleDate
}
}
value = maxInt64(value, valueForEntry(now, scheduledTime, schedule.Duration(), scheduledEndTime, scalingWindowDuration, rampSteps, schedule.Value))
startTime, endTime, err := scheduledscaling.ScheduleStartEnd(now, schedule, defaultTimeZone)
if err != nil {
return nil, err
}
value = maxInt64(value, valueForEntry(now, startTime, endTime, scalingWindowDuration, rampSteps, schedule.Value))
}
return []CollectedMetric{
@ -358,27 +260,17 @@ func calculateMetrics(spec v1.ScalingScheduleSpec, defaultScalingWindow time.Dur
}, nil
}
func valueForEntry(timestamp time.Time, startTime time.Time, entryDuration time.Duration, scheduledEndTime time.Time, scalingWindowDuration time.Duration, rampSteps int, value int64) int64 {
func valueForEntry(timestamp time.Time, startTime time.Time, endTime time.Time, scalingWindowDuration time.Duration, rampSteps int, value int64) int64 {
scaleUpStart := startTime.Add(-scalingWindowDuration)
var endTime time.Time
// Use either the defined end time/date or the start time/date + the
// duration, whichever is longer.
if startTime.Add(entryDuration).Before(scheduledEndTime) {
endTime = scheduledEndTime
} else {
endTime = startTime.Add(entryDuration)
}
scaleUpEnd := endTime.Add(scalingWindowDuration)
if between(timestamp, startTime, endTime) {
if scheduledscaling.Between(timestamp, startTime, endTime) {
return value
}
if between(timestamp, scaleUpStart, startTime) {
if scheduledscaling.Between(timestamp, scaleUpStart, startTime) {
return scaledValue(timestamp, scaleUpStart, scalingWindowDuration, rampSteps, value)
}
if between(timestamp, endTime, scaleUpEnd) {
if scheduledscaling.Between(timestamp, endTime, scaleUpEnd) {
return scaledValue(scaleUpEnd, timestamp, scalingWindowDuration, rampSteps, value)
}
return 0
@ -400,13 +292,6 @@ func scaledValue(timestamp time.Time, startTime time.Time, scalingWindowDuration
return int64(math.Floor(requiredPercentage*steps) * (float64(value) / steps))
}
func between(timestamp, start, end time.Time) bool {
if timestamp.Before(start) {
return false
}
return timestamp.Before(end)
}
func maxInt64(i1, i2 int64) int64 {
if i1 > i2 {
return i1

22
pkg/collector/scaling_schedule_collector_test.go
View File

@ -8,6 +8,7 @@ import (
"github.com/stretchr/testify/require"
v1 "github.com/zalando-incubator/kube-metrics-adapter/pkg/apis/zalando.org/v1"
scheduledscaling "github.com/zalando-incubator/kube-metrics-adapter/pkg/controller/scheduledscaling"
autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@ -16,6 +17,7 @@ const (
hHMMFormat = "15:04"
defaultScalingWindowDuration = 1 * time.Minute
defaultRampSteps = 10
defaultTimeZone = "Europe/Berlin"
)
type schedule struct {
@ -247,7 +249,7 @@ func TestScalingScheduleCollector(t *testing.T) {
value: 100,
},
},
err: ErrInvalidScheduleDate,
err: scheduledscaling.ErrInvalidScheduleDate,
},
{
msg: "Return error for one time config end date not in RFC3339 format",
@ -260,7 +262,7 @@ func TestScalingScheduleCollector(t *testing.T) {
value: 100,
},
},
err: ErrInvalidScheduleDate,
err: scheduledscaling.ErrInvalidScheduleDate,
},
{
msg: "Return the right value for two one time config",
@ -413,7 +415,7 @@ func TestScalingScheduleCollector(t *testing.T) {
days: []v1.ScheduleDay{nowWeekday},
},
},
err: ErrInvalidScheduleStartTime,
err: scheduledscaling.ErrInvalidScheduleStartTime,
},
{
msg: "Return the right value for a repeating schedule in the right timezone even in the day after it",
@ -598,15 +600,15 @@ func TestScalingScheduleCollector(t *testing.T) {
schedules := getSchedules(tc.schedules)
store := newMockStore(scalingScheduleName, namespace, tc.scalingWindowDurationMinutes, schedules)
plugin, err := NewScalingScheduleCollectorPlugin(store, now, defaultScalingWindowDuration, rampSteps)
plugin, err := NewScalingScheduleCollectorPlugin(store, now, defaultScalingWindowDuration, defaultTimeZone, rampSteps)
require.NoError(t, err)
clusterStore := newClusterMockStore(scalingScheduleName, tc.scalingWindowDurationMinutes, schedules)
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(clusterStore, now, defaultScalingWindowDuration, rampSteps)
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(clusterStore, now, defaultScalingWindowDuration, defaultTimeZone, rampSteps)
require.NoError(t, err)
clusterStoreFirstRun := newClusterMockStoreFirstRun(scalingScheduleName, tc.scalingWindowDurationMinutes, schedules)
clusterPluginFirstRun, err := NewClusterScalingScheduleCollectorPlugin(clusterStoreFirstRun, now, defaultScalingWindowDuration, rampSteps)
clusterPluginFirstRun, err := NewClusterScalingScheduleCollectorPlugin(clusterStoreFirstRun, now, defaultScalingWindowDuration, defaultTimeZone, rampSteps)
require.NoError(t, err)
hpa := makeScalingScheduleHPA(namespace, scalingScheduleName)
@ -674,14 +676,14 @@ func TestScalingScheduleObjectNotPresentReturnsError(t *testing.T) {
make(map[string]interface{}),
getByKeyFn,
}
plugin, err := NewScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultRampSteps)
plugin, err := NewScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultTimeZone, defaultRampSteps)
require.NoError(t, err)
clusterStore := mockStore{
make(map[string]interface{}),
getByKeyFn,
}
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(clusterStore, time.Now, defaultScalingWindowDuration, defaultRampSteps)
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(clusterStore, time.Now, defaultScalingWindowDuration, defaultTimeZone, defaultRampSteps)
require.NoError(t, err)
hpa := makeScalingScheduleHPA("namespace", "scalingScheduleName")
@ -736,10 +738,10 @@ func TestReturnsErrorWhenStoreDoes(t *testing.T) {
},
}
plugin, err := NewScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultRampSteps)
plugin, err := NewScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultTimeZone, defaultRampSteps)
require.NoError(t, err)
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultRampSteps)
clusterPlugin, err := NewClusterScalingScheduleCollectorPlugin(store, time.Now, defaultScalingWindowDuration, defaultTimeZone, defaultRampSteps)
require.NoError(t, err)
hpa := makeScalingScheduleHPA("namespace", "scalingScheduleName")

142
pkg/controller/scheduledscaling/scheduled_scaling.go Normal file
View File

@ -0,0 +1,142 @@
package scheduledscaling
import (
"errors"
"fmt"
"time"
v1 "github.com/zalando-incubator/kube-metrics-adapter/pkg/apis/zalando.org/v1"
)
const (
// The format used by v1.SchedulePeriod.StartTime. 15:04 are
// the defined reference time in time.Format.
hourColonMinuteLayout = "15:04"
)
var days = map[v1.ScheduleDay]time.Weekday{
v1.SundaySchedule: time.Sunday,
v1.MondaySchedule: time.Monday,
v1.TuesdaySchedule: time.Tuesday,
v1.WednesdaySchedule: time.Wednesday,
v1.ThursdaySchedule: time.Thursday,
v1.FridaySchedule: time.Friday,
v1.SaturdaySchedule: time.Saturday,
}
var (
// ErrNotScalingScheduleFound is returned when a item returned from
// the ScalingScheduleCollectorPlugin.store was expected to
// be an ScalingSchedule but the type assertion failed.
ErrNotScalingScheduleFound = errors.New("error converting returned object to ScalingSchedule")
// ErrInvalidScheduleDate is returned when the v1.ScheduleDate is
// not a valid RFC3339 date. It shouldn't happen since the
// validation is done by the CRD.
ErrInvalidScheduleDate = errors.New("could not parse the specified schedule date, format is not RFC3339")
// ErrInvalidScheduleStartTime is returned when the
// v1.SchedulePeriod.StartTime is not in the format specified by
// hourColonMinuteLayout. It shouldn't happen since the validation
// is done by the CRD.
ErrInvalidScheduleStartTime = errors.New("could not parse the specified schedule period start time, format is not HH:MM")
)
// Now is the function that returns a time.Time object representing the
// current moment. Its main implementation is the time.Now func in the
// std lib. It's used mainly for test/mock purposes.
type now func() time.Time
func ScheduleStartEnd(now time.Time, schedule v1.Schedule, defaultTimeZone string) (time.Time, time.Time, error) {
var startTime, endTime time.Time
switch schedule.Type {
case v1.RepeatingSchedule:
location, err := time.LoadLocation(schedule.Period.Timezone)
if schedule.Period.Timezone == "" || err != nil {
location, err = time.LoadLocation(defaultTimeZone)
if err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("unexpected error loading default location: %s", err.Error())
}
}
nowInLocation := now.In(location)
weekday := nowInLocation.Weekday()
for _, day := range schedule.Period.Days {
if days[day] == weekday {
parsedStartTime, err := time.Parse(hourColonMinuteLayout, schedule.Period.StartTime)
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidScheduleStartTime
}
startTime = time.Date(
// v1.SchedulePeriod.StartTime can't define the
// year, month or day, so we compute it as the
// current date in the configured location.
nowInLocation.Year(),
nowInLocation.Month(),
nowInLocation.Day(),
// Hours and minute are configured in the
// v1.SchedulePeriod.StartTime.
parsedStartTime.Hour(),
parsedStartTime.Minute(),
parsedStartTime.Second(),
parsedStartTime.Nanosecond(),
location,
)
// If no end time was provided, set it to equal the start time
if schedule.Period.EndTime == "" {
endTime = startTime
} else {
parsedEndTime, err := time.Parse(hourColonMinuteLayout, schedule.Period.EndTime)
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidScheduleDate
}
endTime = time.Date(
// v1.SchedulePeriod.StartTime can't define the
// year, month or day, so we compute it as the
// current date in the configured location.
nowInLocation.Year(),
nowInLocation.Month(),
nowInLocation.Day(),
// Hours and minute are configured in the
// v1.SchedulePeriod.StartTime.
parsedEndTime.Hour(),
parsedEndTime.Minute(),
parsedEndTime.Second(),
parsedEndTime.Nanosecond(),
location,
)
}
}
}
case v1.OneTimeSchedule:
var err error
startTime, err = time.Parse(time.RFC3339, string(*schedule.Date))
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidScheduleDate
}
// If no end time was provided, set it to equal the start time
if schedule.EndDate == nil || string(*schedule.EndDate) == "" {
endTime = startTime
} else {
endTime, err = time.Parse(time.RFC3339, string(*schedule.EndDate))
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidScheduleDate
}
}
}
// Use either the defined end time/date or the start time/date + the
// duration, whichever is longer.
if startTime.Add(schedule.Duration()).After(endTime) {
endTime = startTime.Add(schedule.Duration())
}
return startTime, endTime, nil
}
func Between(timestamp, start, end time.Time) bool {
if timestamp.Before(start) {
return false
}
return timestamp.Before(end)
}

7
pkg/server/start.go
View File

@ -130,6 +130,7 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
"whether to enable time-based ScalingSchedule metrics")
flags.DurationVar(&o.DefaultScheduledScalingWindow, "scaling-schedule-default-scaling-window", 10*time.Minute, "Default rampup and rampdown window duration for ScalingSchedules")
flags.IntVar(&o.RampSteps, "scaling-schedule-ramp-steps", 10, "Number of steps used to rampup and rampdown ScalingSchedules. It's used to guarantee won't avoid reaching the max scaling due to the 10% minimum change rule.")
flags.StringVar(&o.DefaultTimeZone, "scaling-schedule-default-time-zone", "Europe/Berlin", "Default time zone to use for ScalingSchedules.")
return cmd
}
@ -295,7 +296,7 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
)
go reflector.Run(ctx.Done())
clusterPlugin, err := collector.NewClusterScalingScheduleCollectorPlugin(clusterScalingSchedulesStore, time.Now, o.DefaultScheduledScalingWindow, o.RampSteps)
clusterPlugin, err := collector.NewClusterScalingScheduleCollectorPlugin(clusterScalingSchedulesStore, time.Now, o.DefaultScheduledScalingWindow, o.DefaultTimeZone, o.RampSteps)
if err != nil {
return fmt.Errorf("unable to create ClusterScalingScheduleCollector plugin: %v", err)
}
@ -304,7 +305,7 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
return fmt.Errorf("failed to register ClusterScalingSchedule object collector plugin: %v", err)
}
plugin, err := collector.NewScalingScheduleCollectorPlugin(scalingSchedulesStore, time.Now, o.DefaultScheduledScalingWindow, o.RampSteps)
plugin, err := collector.NewScalingScheduleCollectorPlugin(scalingSchedulesStore, time.Now, o.DefaultScheduledScalingWindow, o.DefaultTimeZone, o.RampSteps)
if err != nil {
return fmt.Errorf("unable to create ScalingScheduleCollector plugin: %v", err)
}
@ -434,4 +435,6 @@ type AdapterServerOptions struct {
DefaultScheduledScalingWindow time.Duration
// Number of steps utilized during the rampup and rampdown for scheduled metrics
RampSteps int
// Default time zone to use for ScalingSchedules.
DefaultTimeZone string
}