Add Optuna based suggestion service

cmd/suggestion/optuna/v1beta1/Dockerfile

@@ -0,0 +1,32 @@
+FROM python:3.6
+
+ENV TARGET_DIR /opt/katib
+ENV SUGGESTION_DIR cmd/suggestion/optuna/v1beta1
+
+RUN if [ "$(uname -m)" = "ppc64le" ] || [ "$(uname -m)" = "aarch64" ]; then \
+    apt-get -y update && \
+    apt-get -y install gfortran libopenblas-dev liblapack-dev && \
+    pip install cython; \
+    fi
+
+RUN GRPC_HEALTH_PROBE_VERSION=v0.3.1 && \
+    if [ "$(uname -m)" = "ppc64le" ]; then \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-ppc64le; \
+    elif [ "$(uname -m)" = "aarch64" ]; then \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-arm64; \
+    else \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64; \
+    fi && \
+    chmod +x /bin/grpc_health_probe
+
+ADD ./pkg/ ${TARGET_DIR}/pkg/
+ADD ./${SUGGESTION_DIR}/ ${TARGET_DIR}/${SUGGESTION_DIR}/
+WORKDIR  ${TARGET_DIR}/${SUGGESTION_DIR}
+RUN pip install --no-cache-dir -r requirements.txt
+
+RUN chgrp -R 0 ${TARGET_DIR} \
+  && chmod -R g+rwX ${TARGET_DIR}
+
+ENV PYTHONPATH ${TARGET_DIR}:${TARGET_DIR}/pkg/apis/manager/v1beta1/python:${TARGET_DIR}/pkg/apis/manager/health/python
+
+ENTRYPOINT ["python", "main.py"]

cmd/suggestion/optuna/v1beta1/main.py

@@ -0,0 +1,42 @@
+# Copyright 2021 The Kubeflow Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import grpc
+import time
+from pkg.apis.manager.v1beta1.python import api_pb2_grpc
+from pkg.apis.manager.health.python import health_pb2_grpc
+from pkg.suggestion.v1beta1.optuna.service import OptunaService
+from concurrent import futures
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+DEFAULT_PORT = "0.0.0.0:6789"
+
+
+def serve():
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+    service = OptunaService()
+    api_pb2_grpc.add_SuggestionServicer_to_server(service, server)
+    health_pb2_grpc.add_HealthServicer_to_server(service, server)
+    server.add_insecure_port(DEFAULT_PORT)
+    print("Listening...")
+    server.start()
+    try:
+        while True:
+            time.sleep(_ONE_DAY_IN_SECONDS)
+    except KeyboardInterrupt:
+        server.stop(0)
+
+
+if __name__ == "__main__":
+    serve()

cmd/suggestion/optuna/v1beta1/requirements.txt

@@ -0,0 +1,4 @@
+grpcio==1.23.0
+protobuf==3.9.1
+googleapis-common-protos==1.6.0
+optuna>=2.8.0

pkg/suggestion/v1beta1/optuna/service.py

@@ -0,0 +1,183 @@
+# Copyright 2021 The Kubeflow Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from collections import defaultdict
+import logging
+import threading
+
+import optuna
+
+from pkg.apis.manager.v1beta1.python import api_pb2
+from pkg.apis.manager.v1beta1.python import api_pb2_grpc
+from pkg.suggestion.v1beta1.internal.constant import INTEGER, DOUBLE, CATEGORICAL, DISCRETE, MAX_GOAL
+from pkg.suggestion.v1beta1.internal.search_space import HyperParameterSearchSpace
+from pkg.suggestion.v1beta1.internal.trial import Trial, Assignment
+from pkg.suggestion.v1beta1.internal.base_health_service import HealthServicer
+
+
+logger = logging.getLogger(__name__)
+
+
+class OptunaService(api_pb2_grpc.SuggestionServicer, HealthServicer):
+
+    def __init__(self):
+        super(OptunaService, self).__init__()
+        self.study = None
+        self.search_space = None
+        self.recorded_trial_names = set()
+        self.assignments_to_optuna_number = defaultdict(list)
+        self.lock = threading.Lock()
+
+    def GetSuggestions(self, request, context):
+        """
+        Main function to provide suggestion.
+        """
+        with self.lock:
+            if self.study is None:
+                self.search_space = HyperParameterSearchSpace.convert(request.experiment)
+                self.study = self._create_study(request.experiment.spec.algorithm, self.search_space)
+
+        trials = Trial.convert(request.trials)
+
+        with self.lock:
+            if len(trials) != 0:
+                self._tell(trials)
+            list_of_assignments = self._ask(request.request_number)
+
+        return api_pb2.GetSuggestionsReply(
+            parameter_assignments=Assignment.generate(list_of_assignments)
+        )
+
+    def _create_study(self, algorithm_spec, search_space):
+        sampler = self._create_sampler(algorithm_spec)
+        direction = "maximize" if search_space.goal == MAX_GOAL else "minimize"
+
+        study = optuna.create_study(sampler=sampler, direction=direction)
+
+        return study
+
+    def _create_sampler(self, algorithm_spec):
+        name = algorithm_spec.algorithm_name
+        settings = {s.name:s.value for s in algorithm_spec.algorithm_settings}
+
+        if name == "tpe" or name == "multivariate-tpe":
+            kwargs = {}
+            for k, v in settings.items():
+                if k == "startup_trials":
+                    kwargs["n_startup_trials"] = int(v)
+                elif k == "ei_candidates":
+                    kwargs["n_ei_candidates"] = int(v)
+                elif k == "random_state":
+                    kwargs["seed"] = int(v)
+                else:
+                    raise ValueError("Unknown name for {}: {}".format(name, k))
+
+            kwargs["multivariate"] = name == "multivariate-tpe"
+
+            sampler = optuna.samplers.TPESampler(**kwargs)
+
+        elif name == "cmaes":
+            kwargs = {}
+            for k, v in settings.items():
+                if k == "restart_strategy":
+                    kwargs["restart_strategy"] = v
+                elif k == "sigma":
+                    kwargs["sigma0"] = float(v)
+                elif k == "random_state":
+                    kwargs["seed"] = int(v)
+                else:
+                    raise ValueError("Unknown name for {}: {}".format(name, k))
+
+            sampler = optuna.samplers.CmaEsSampler(**kwargs)
+
+        elif name == "random":
+            kwargs = {}
+            for k, v in settings.items():
+                if k == "random_state":
+                    kwargs["seed"] = int(v)
+                else:
+                    raise ValueError("Unknown name for {}: {}".format(name, k))
+
+            sampler = optuna.samplers.RandomSampler(**kwargs)
+
+        else:
+            raise ValueError("Unknown algorithm name: {}".format(name))
+
+        return sampler
+
+    def _ask(self, request_number):
+        list_of_assignments = []
+        for _ in range(request_number):
+            optuna_trial = self.study.ask(fixed_distributions=self._get_optuna_serch_space())
+
+            assignments = [Assignment(k,v) for k,v in optuna_trial.params.items()]
+            list_of_assignments.append(assignments)
+
+            assignments_key = self._get_assignments_key(assignments)
+            self.assignments_to_optuna_number[assignments_key].append(optuna_trial.number)
+
+        return list_of_assignments
+
+    def _tell(self, trials):
+        for trial in trials:
+            if trial.name not in self.recorded_trial_names:
+                self.recorded_trial_names.add(trial.name)
+
+                value = float(trial.target_metric.value)
+                assignments_key = self._get_assignments_key(trial.assignments)
+                optuna_trial_numbers = self.assignments_to_optuna_number[assignments_key]
+
+                if len(optuna_trial_numbers) != 0:
+                    # The trial has been suggested by the Optuna study.
+                    # The objective value is reported using study.tell() with the corresponding trial number.
+                    trial_number = optuna_trial_numbers.pop(0)
+                    self.study.tell(trial_number, value)
+                else:
+                    # The trial has not been suggested by the Optuna study.
+                    # A new trial object is created and reported using study.add_trial() with the assignments and the search space.
+                    optuna_trial = optuna.create_trial(
+                        params={a.name:self._get_casted_assignment_value(a) for a in trial.assignments},
+                        distributions=self._get_optuna_serch_space(),
+                        value=value,
+                    )
+                    self.study.add_trial(optuna_trial)
+
+    def _get_assignments_key(self, assignments):
+        assignments = sorted(assignments, key=lambda a: a.name)
+        assignments_str = [str(a) for a in assignments]
+        return ",".join(assignments_str)
+
+    def _get_optuna_serch_space(self):
+        search_space = {}
+        for param in self.search_space.params:
+            if param.type == INTEGER:
+                search_space[param.name] = optuna.distributions.IntUniformDistribution(int(param.min), int(param.max))
+            elif param.type == DOUBLE:
+                search_space[param.name] = optuna.distributions.UniformDistribution(float(param.min), float(param.max))
+            elif param.type == CATEGORICAL or param.type == DISCRETE:
+                search_space[param.name] = optuna.distributions.CategoricalDistribution(param.list)
+        return search_space
+
+    def _get_casted_assignment_value(self, assignment):
+        for param in self.search_space.params:
+            if param.name == assignment.name:
+                if param.type == INTEGER:
+                    return int(assignment.value)
+                elif param.type == DOUBLE:
+                    return float(assignment.value)
+                elif param.type == CATEGORICAL or param.type == DISCRETE:
+                    return assignment.value
+                else:
+                    raise ValueError("Unknown parameter type: {}".format(param.type))
+        raise ValueError("Parameter not found in the search space: {}".format(param.name))

scripts/v1beta1/build.sh

@@ -85,6 +85,9 @@ docker build -t ${REGISTRY}/suggestion-skopt:${TAG} -f ${CMD_PREFIX}/suggestion/
 echo -e "\nBuilding goptuna suggestion...\n"
 docker build -t ${REGISTRY}/suggestion-goptuna:${TAG} -f ${CMD_PREFIX}/suggestion/goptuna/${VERSION}/Dockerfile .
 
+echo -e "\nBuilding optuna suggestion...\n"
+docker build -t ${REGISTRY}/suggestion-optuna:${TAG} -f ${CMD_PREFIX}/suggestion/optuna/${VERSION}/Dockerfile .
+
 echo -e "\nBuilding ENAS suggestion...\n"
 if [ $MACHINE_ARCH == "aarch64" ]; then
     docker build -t ${REGISTRY}/suggestion-enas:${TAG} -f ${CMD_PREFIX}/suggestion/nas/enas/${VERSION}/Dockerfile.aarch64 .

test/scripts/v1beta1/python-tests.sh

@@ -25,6 +25,7 @@ pip install -r test/suggestion/v1beta1/test_requirements.txt
 pip install -r cmd/suggestion/chocolate/v1beta1/requirements.txt
 pip install -r cmd/suggestion/hyperopt/v1beta1/requirements.txt
 pip install -r cmd/suggestion/skopt/v1beta1/requirements.txt
+pip install -r cmd/suggestion/optuna/v1beta1/requirements.txt
 pip install -r cmd/suggestion/nas/enas/v1beta1/requirements.txt
 pip install -r cmd/suggestion/hyperband/v1beta1/requirements.txt
 pip install -r cmd/suggestion/nas/darts/v1beta1/requirements.txt