-
Notifications
You must be signed in to change notification settings - Fork 575
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Setup a testing endpoint for Rightsizing (#5)
- Loading branch information
1 parent
a875b17
commit b382625
Showing
11 changed files
with
398 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
84 changes: 84 additions & 0 deletions
84
...src/main/java/com/linkedin/kafka/cruisecontrol/servlet/handler/sync/RightsizeRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| /* | ||
| * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. | ||
| */ | ||
|
|
||
| package com.linkedin.kafka.cruisecontrol.servlet.handler.sync; | ||
|
|
||
| import com.linkedin.kafka.cruisecontrol.KafkaCruiseControl; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionRecommendation; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionStatus; | ||
| import com.linkedin.kafka.cruisecontrol.common.Utils; | ||
| import com.linkedin.kafka.cruisecontrol.config.KafkaCruiseControlConfig; | ||
| import com.linkedin.kafka.cruisecontrol.config.constants.AnomalyDetectorConfig; | ||
| import com.linkedin.kafka.cruisecontrol.detector.Provisioner; | ||
| import com.linkedin.kafka.cruisecontrol.detector.ProvisionerState; | ||
| import com.linkedin.kafka.cruisecontrol.detector.RightsizeOptions; | ||
| import com.linkedin.kafka.cruisecontrol.servlet.parameters.RightsizeParameters; | ||
| import com.linkedin.kafka.cruisecontrol.servlet.response.RightsizeResult; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.function.Supplier; | ||
|
|
||
| import static com.linkedin.cruisecontrol.common.utils.Utils.validateNotNull; | ||
| import static com.linkedin.kafka.cruisecontrol.detector.AnomalyDetectorUtils.KAFKA_CRUISE_CONTROL_OBJECT_CONFIG; | ||
| import static com.linkedin.kafka.cruisecontrol.servlet.parameters.ParameterUtils.RIGHTSIZE_PARAMETER_OBJECT_CONFIG; | ||
|
|
||
|
|
||
| public class RightsizeRequest extends AbstractSyncRequest { | ||
| private static final String RECOMMENDER_UP = "Recommender-Under-Provisioned"; | ||
| protected KafkaCruiseControl _kafkaCruiseControl; | ||
| protected RightsizeParameters _parameters; | ||
|
|
||
| public RightsizeRequest() { | ||
| super(); | ||
| } | ||
|
|
||
| @Override | ||
| protected RightsizeResult handle() { | ||
| KafkaCruiseControlConfig config = _kafkaCruiseControl.config(); | ||
| Map<String, Object> overrideConfigs; | ||
| overrideConfigs = Collections.singletonMap(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, _kafkaCruiseControl); | ||
| Provisioner provisioner = config.getConfiguredInstance(AnomalyDetectorConfig.PROVISIONER_CLASS_CONFIG, | ||
| Provisioner.class, | ||
| overrideConfigs); | ||
| Supplier<Set<String>> topicNameSupplier = () -> _kafkaCruiseControl.kafkaCluster().topics(); | ||
| Set<String> topicNamesMatchedWithPattern = Utils.getTopicNamesMatchedWithPattern(_parameters.topic(), topicNameSupplier); | ||
| String topicName; | ||
| if (topicNamesMatchedWithPattern.size() > 1) { | ||
| throw new IllegalArgumentException(String.format("The RightsizeEndpoint does not support provisioning for multiple topics {%s}.", | ||
| String.join(" ,", topicNamesMatchedWithPattern))); | ||
| } else { | ||
| topicName = topicNamesMatchedWithPattern.iterator().next(); | ||
| } | ||
| ProvisionRecommendation recommendation = | ||
| new ProvisionRecommendation.Builder(ProvisionStatus.UNDER_PROVISIONED).numBrokers(_parameters.numBrokersToAdd()) | ||
| .numPartitions(_parameters.partitionCount()) | ||
| .topic(topicName) | ||
| .build(); | ||
| Map<String, ProvisionRecommendation> provisionRecommendation; | ||
| provisionRecommendation = Collections.singletonMap(RECOMMENDER_UP, recommendation); | ||
|
|
||
| ProvisionerState provisionerState = provisioner.rightsize(provisionRecommendation, new RightsizeOptions()); | ||
|
|
||
| return new RightsizeResult(_parameters.numBrokersToAdd(), _parameters.partitionCount(), topicName, provisionerState, config); | ||
| } | ||
|
|
||
| @Override | ||
| public RightsizeParameters parameters() { | ||
| return _parameters; | ||
| } | ||
|
|
||
| @Override | ||
| public String name() { | ||
| return RightsizeRequest.class.getSimpleName(); | ||
| } | ||
|
|
||
| @Override | ||
| public void configure(Map<String, ?> configs) { | ||
| super.configure(configs); | ||
| _kafkaCruiseControl = _servlet.asyncKafkaCruiseControl(); | ||
| _parameters = (RightsizeParameters) validateNotNull(configs.get(RIGHTSIZE_PARAMETER_OBJECT_CONFIG), | ||
| "Parameter configuration is missing from the request."); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
...rc/main/java/com/linkedin/kafka/cruisecontrol/servlet/parameters/RightsizeParameters.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| /* | ||
| * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. | ||
| */ | ||
|
|
||
| package com.linkedin.kafka.cruisecontrol.servlet.parameters; | ||
|
|
||
| import java.io.UnsupportedEncodingException; | ||
| import java.util.Collections; | ||
| import java.util.SortedSet; | ||
| import java.util.TreeSet; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| import static com.linkedin.kafka.cruisecontrol.servlet.parameters.ParameterUtils.NUM_BROKERS_TO_ADD; | ||
| import static com.linkedin.kafka.cruisecontrol.servlet.parameters.ParameterUtils.PARTITION_COUNT; | ||
| import static com.linkedin.kafka.cruisecontrol.servlet.parameters.ParameterUtils.TOPIC_PARAM; | ||
|
|
||
|
|
||
| public class RightsizeParameters extends AbstractParameters { | ||
| protected static final SortedSet<String> CASE_INSENSITIVE_PARAMETER_NAMES; | ||
| static { | ||
| SortedSet<String> validParameterNames = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); | ||
| validParameterNames.add(NUM_BROKERS_TO_ADD); | ||
| validParameterNames.add(PARTITION_COUNT); | ||
| validParameterNames.add(TOPIC_PARAM); | ||
| validParameterNames.addAll(KafkaOptimizationParameters.CASE_INSENSITIVE_PARAMETER_NAMES); | ||
| CASE_INSENSITIVE_PARAMETER_NAMES = Collections.unmodifiableSortedSet(validParameterNames); | ||
| } | ||
| protected int _numBrokersToAdd; | ||
| protected int _partitionCount; | ||
| protected Pattern _topic; | ||
|
|
||
| public RightsizeParameters() { | ||
| super(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void initParameters() throws UnsupportedEncodingException { | ||
| super.initParameters(); | ||
| _numBrokersToAdd = ParameterUtils.numBrokersToAdd(_request); | ||
| _partitionCount = ParameterUtils.partitionCount(_request); | ||
| _topic = ParameterUtils.topic(_request); | ||
| } | ||
|
|
||
| public int numBrokersToAdd() { | ||
| return _numBrokersToAdd; | ||
| } | ||
|
|
||
| public int partitionCount() { | ||
| return _partitionCount; | ||
| } | ||
|
|
||
| public Pattern topic() { | ||
| return _topic; | ||
| } | ||
|
|
||
| @Override | ||
| public SortedSet<String> caseInsensitiveParameterNames() { | ||
| return CASE_INSENSITIVE_PARAMETER_NAMES; | ||
| } | ||
| } |
73 changes: 73 additions & 0 deletions
73
...trol/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/response/RightsizeResult.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| /* | ||
| * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. | ||
| */ | ||
|
|
||
| package com.linkedin.kafka.cruisecontrol.servlet.response; | ||
|
|
||
| import com.google.gson.Gson; | ||
| import com.linkedin.cruisecontrol.servlet.parameters.CruiseControlParameters; | ||
| import com.linkedin.kafka.cruisecontrol.config.KafkaCruiseControlConfig; | ||
| import com.linkedin.kafka.cruisecontrol.detector.ProvisionerState; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
|
|
||
| @JsonResponseClass | ||
| public class RightsizeResult extends AbstractCruiseControlResponse { | ||
| @JsonResponseField | ||
| protected static final String NUM_BROKERS_TO_ADD = "num_brokers_to_add"; | ||
| @JsonResponseField | ||
| protected static final String PARTITION_COUNT = "partition_count"; | ||
| @JsonResponseField | ||
| protected static final String TOPIC = "topic"; | ||
| @JsonResponseField | ||
| protected static final String PROVISION_STATE = "provision_state"; | ||
| @JsonResponseField | ||
| protected static final String PROVISION_SUMMARY = "provision_summary"; | ||
|
|
||
| protected final int _numBrokersToAdd; | ||
| protected final int _partitionCount; | ||
| protected String _topic; | ||
| protected String _provisionState; | ||
| protected String _provisionSummary; | ||
|
|
||
| public RightsizeResult(int numBrokersToAdd, | ||
| int partitionCount, | ||
| String topic, | ||
| ProvisionerState provisionerState, | ||
| KafkaCruiseControlConfig config) { | ||
| super(config); | ||
|
|
||
| _numBrokersToAdd = numBrokersToAdd; | ||
| _partitionCount = partitionCount; | ||
| _topic = topic; | ||
| _provisionState = provisionerState.state().toString(); | ||
| _provisionSummary = provisionerState.summary(); | ||
| } | ||
|
|
||
| protected String getJSONString() { | ||
| Map<String, Object> jsonStructure = new HashMap<>(5); | ||
| jsonStructure.put(NUM_BROKERS_TO_ADD, _numBrokersToAdd); | ||
| jsonStructure.put(PARTITION_COUNT, _partitionCount); | ||
| if (_topic != null && !_topic.isEmpty()) { | ||
| jsonStructure.put(TOPIC, _topic); | ||
| } | ||
| if (_provisionState != null && !_provisionState.isEmpty()) { | ||
| jsonStructure.put(PROVISION_STATE, _provisionState); | ||
| } | ||
| if (_provisionSummary != null && !_provisionSummary.isEmpty()) { | ||
| jsonStructure.put(PROVISION_SUMMARY, _provisionSummary); | ||
| } | ||
| Gson gson = new Gson(); | ||
| return gson.toJson(jsonStructure); | ||
| } | ||
|
|
||
| @Override | ||
| protected void discardIrrelevantAndCacheRelevant(CruiseControlParameters parameters) { | ||
| _cachedResponse = getJSONString(); | ||
| // Discard irrelevant response. | ||
| _topic = null; | ||
| _provisionState = null; | ||
| _provisionSummary = null; | ||
| } | ||
| } |
Oops, something went wrong.