Skip to content

Commit

Permalink
Change HPA to account for Kubernetes tolerances (#681)
Browse files Browse the repository at this point in the history 
Co-authored-by: Bryan Burkholder <[email protected]>
  • Loading branch information
@bryanlb
bryanlb and bryanlb authored Sep 22, 2023
1 parent d2c7dcd commit 3818a36
Show file tree
Hide file tree
Showing 3 changed files with 62 additions and 41 deletions.
69 changes: 47 additions & 22 deletions kaldb/src/main/java/com/slack/kaldb/clusterManager/ClusterHpaMetricService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,9 @@
public class ClusterHpaMetricService extends AbstractScheduledService {
private static final Logger LOG = LoggerFactory.getLogger(ClusterHpaMetricService.class);

// todo - consider making over-provision and lock duration configurable values
protected int CACHE_SLOT_OVER_PROVISION = 1000;
// todo - consider making HPA_TOLERANCE and CACHE_SCALEDOWN_LOCK configurable
// https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/#algorithm-details
private static final double HPA_TOLERANCE = 0.1;
protected Duration CACHE_SCALEDOWN_LOCK = Duration.of(15, ChronoUnit.MINUTES);

private final ReplicaMetadataStore replicaMetadataStore;
Expand Down Expand Up @@ -103,34 +104,58 @@ private void publishCacheHpaMetrics() {
replicaMetadataStore.listSync().stream()
.filter(replicaMetadata -> replicaMetadata.getReplicaSet().equals(replicaSet))
.count();
double rawDemandFactor =
(double) (totalReplicaDemand + CACHE_SLOT_OVER_PROVISION) / (totalCacheSlotCapacity + 1);
double demandFactor = (double) Math.round(rawDemandFactor * 100) / 100;
LOG.info(
"Cache autoscaler for replicaSet '{}' calculated a demandFactor of '{}' - totalReplicaDemand: '{}', cacheSlotOverProvision: '{}', totalCacheSlotCapacity: '{}'",
replicaSet,
demandFactor,
totalReplicaDemand,
CACHE_SLOT_OVER_PROVISION,
totalCacheSlotCapacity);

if (demandFactor >= 1.0) {
// Publish a scale-up metric
double demandFactor = calculateDemandFactor(totalCacheSlotCapacity, totalReplicaDemand);
String action;
if (demandFactor > 1) {
// scale-up
if (demandFactor < (1 + HPA_TOLERANCE)) {
// scale-up required, but still within the HPA tolerance
// we need to ensure the scale-up is at least triggering the HPA
demandFactor = demandFactor + HPA_TOLERANCE;
}
action = "scale-up";
persistCacheConfig(replicaSet, demandFactor);
LOG.debug("Publishing scale-up request for replicaset '{}'", replicaSet);
} else {
} else if (demandFactor < (1 - HPA_TOLERANCE)) {
// scale-down required
if (tryCacheReplicasetLock(replicaSet)) {
// Publish a scale-down metric
action = "scale-down";
persistCacheConfig(replicaSet, demandFactor);
LOG.debug("Acquired scale-down lock for replicaSet '{}'", replicaSet);
} else {
// Publish a no-op scale metric (0) to disable the HPA from applying
// https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/#implicit-maintenance-mode-deactivation
persistCacheConfig(replicaSet, 0.0);
LOG.debug("Unable to acquire scale-down lock for replicaSet '{}'", replicaSet);
// couldn't get exclusive lock, no-op
action = "pending-scale-down";
persistCacheConfig(replicaSet, 1.0);
}
} else {
// over-provisioned, but within HPA tolerance
action = "no-op";
persistCacheConfig(replicaSet, demandFactor);
}

LOG.info(
"Cache autoscaler for replicaset '{}' took action '{}', demandFactor: '{}', totalReplicaDemand: '{}', totalCacheSlotCapacity: '{}'",
replicaSet,
action,
demandFactor,
totalReplicaDemand,
totalCacheSlotCapacity);
}
}

private static double calculateDemandFactor(
long totalCacheSlotCapacity, long totalReplicaDemand) {
if (totalCacheSlotCapacity == 0) {
// we have no provisioned capacity, so cannot determine a value
// this should never happen unless the user misconfigured the HPA with a minimum instance
// count of 0
LOG.error(
"No cache slot capacity is detected, this indicates a misconfiguration of the HPA minimum instance count which must be at least 1");
return 1;
}
// demand factor will be < 1 indicating a scale-down demand, and > 1 indicating a scale-up
double rawDemandFactor = (double) (totalReplicaDemand) / (totalCacheSlotCapacity);
// round to 2 decimals
return (double) Math.round(rawDemandFactor * 100) / 100;
}

/** Updates or inserts an (ephemeral) HPA metric for the cache nodes. This is NOT threadsafe. */
Expand Down
5 changes: 2 additions & 3 deletions kaldb/src/main/java/com/slack/kaldb/server/HpaMetricPublisherService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,8 @@ private KaldbMetadataStoreChangeListener<HpaMetricMetadata> changeListener() {
if (metric.isPresent()) {
return metric.get().getValue();
} else {
// store no longer has this metric - report a 0
// https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/#implicit-maintenance-mode-deactivation
return 0;
// store no longer has this metric - report a nominal value 1
return 1;
}
});
}
Expand Down
29 changes: 13 additions & 16 deletions kaldb/src/test/java/com/slack/kaldb/clusterManager/ClusterHpaMetricServiceTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,6 @@ void oneReplicasetScaledown() {
ClusterHpaMetricService clusterHpaMetricService =
new ClusterHpaMetricService(
replicaMetadataStore, cacheSlotMetadataStore, hpaMetricMetadataStore);
clusterHpaMetricService.CACHE_SLOT_OVER_PROVISION = 1;

when(replicaMetadataStore.listSync())
.thenReturn(
Expand Down Expand Up @@ -167,10 +166,10 @@ void oneReplicasetScaledown() {
.findFirst()
.get();

// 2 replicas, 3 slots (2+1)/(3+1)
assertThat(rep1Metadata.getValue()).isEqualTo(0.75);
// 2 replicas, 3 slots
assertThat(rep1Metadata.getValue()).isEqualTo(0.67);

// 1 replica, 1 slot (1+1)/(1+1)
// 1 replica, 1 slot
assertThat(rep2Metadata.getValue()).isEqualTo(1.0);
}

Expand All @@ -180,7 +179,6 @@ void twoReplicasetScaledown() {
ClusterHpaMetricService clusterHpaMetricService =
new ClusterHpaMetricService(
replicaMetadataStore, cacheSlotMetadataStore, hpaMetricMetadataStore);
clusterHpaMetricService.CACHE_SLOT_OVER_PROVISION = 1;

when(replicaMetadataStore.listSync())
.thenReturn(
Expand Down Expand Up @@ -251,13 +249,13 @@ void twoReplicasetScaledown() {
.findFirst()
.get();

// only one should get a lock, the other should be (1 + 1) / (2 + 1))
if (rep1Metadata.getValue().equals(0.0)) {
assertThat(rep1Metadata.getValue()).isEqualTo(0.0);
assertThat(rep2Metadata.getValue()).isEqualTo(0.67);
// only one should get a lock, the other should be 1 replica, 2 slots
if (rep1Metadata.getValue().equals(1.0)) {
assertThat(rep1Metadata.getValue()).isEqualTo(1.0);
assertThat(rep2Metadata.getValue()).isEqualTo(0.5);
} else {
assertThat(rep1Metadata.getValue()).isEqualTo(0.67);
assertThat(rep2Metadata.getValue()).isEqualTo(0.0);
assertThat(rep1Metadata.getValue()).isEqualTo(0.5);
assertThat(rep2Metadata.getValue()).isEqualTo(1.0);
}
}

Expand All @@ -267,7 +265,6 @@ void twoReplicasetScaleup() {
ClusterHpaMetricService clusterHpaMetricService =
new ClusterHpaMetricService(
replicaMetadataStore, cacheSlotMetadataStore, hpaMetricMetadataStore);
clusterHpaMetricService.CACHE_SLOT_OVER_PROVISION = 1;

when(replicaMetadataStore.listSync())
.thenReturn(
Expand Down Expand Up @@ -318,10 +315,10 @@ void twoReplicasetScaleup() {
.findFirst()
.get();

// (2 + 1) / (1 + 1)
assertThat(rep1Metadata.getValue()).isEqualTo(1.5);
// 2 replicas, 1 slot
assertThat(rep1Metadata.getValue()).isEqualTo(2);

// (2 + 1) / 1
assertThat(rep2Metadata.getValue()).isEqualTo(3.0);
// 2 replicas, 0 slots (will log an error and return a default no-op)
assertThat(rep2Metadata.getValue()).isEqualTo(1);
}
}

0 comments on commit 3818a36

Please sign in to comment.