/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

package org.apache.pulsar.functions.runtime.kubernetes;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.Gson;
import com.google.protobuf.Empty;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.apis.AppsV1Api;
import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.custom.Quantity;
import io.kubernetes.client.openapi.models.V1Container;
import io.kubernetes.client.openapi.models.V1ContainerPort;
import io.kubernetes.client.openapi.models.V1DeleteOptions;
import io.kubernetes.client.openapi.models.V1EnvVar;
import io.kubernetes.client.openapi.models.V1EnvVarSource;
import io.kubernetes.client.openapi.models.V1LabelSelector;
import io.kubernetes.client.openapi.models.V1ObjectFieldSelector;
import io.kubernetes.client.openapi.models.V1ObjectMeta;
import io.kubernetes.client.openapi.models.V1PodList;
import io.kubernetes.client.openapi.models.V1PodSpec;
import io.kubernetes.client.openapi.models.V1PodTemplateSpec;
import io.kubernetes.client.openapi.models.V1ResourceRequirements;
import io.kubernetes.client.openapi.models.V1Service;
import io.kubernetes.client.openapi.models.V1ServicePort;
import io.kubernetes.client.openapi.models.V1ServiceSpec;
import io.kubernetes.client.openapi.models.V1StatefulSet;
import io.kubernetes.client.openapi.models.V1StatefulSetSpec;
import io.kubernetes.client.openapi.models.V1Toleration;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Response;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.functions.auth.KubernetesFunctionAuthProvider;
import org.apache.pulsar.common.functions.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
import org.apache.pulsar.functions.runtime.Runtime;
import org.apache.pulsar.functions.runtime.RuntimeUtils;
import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
import org.apache.pulsar.functions.utils.Actions;
import org.apache.pulsar.functions.utils.FunctionCommon;
import org.apache.pulsar.packages.management.core.common.PackageType;

import java.io.IOException;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static java.net.HttpURLConnection.HTTP_CONFLICT;
import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.commons.lang3.StringUtils.left;
import static org.apache.pulsar.functions.auth.FunctionAuthUtils.getFunctionAuthData;
import static org.apache.pulsar.functions.utils.FunctionCommon.roundDecimal;

/**
 * Kubernetes based runtime for running functions.
 * This runtime provides the usual methods to start/stop/getfunctionstatus
 * interfaces to control the kubernetes job running function.
 * We first create a headless service and then a statefulset for starting function pods
 * Each function instance runs as a pod itself. The reason using statefulset as opposed
 * to a regular deployment is that functions require a unique instance_id for each instance.
 * The service abstraction is used for getting functionstatus.
 */
@Slf4j
@VisibleForTesting
public class KubernetesRuntime implements Runtime {

    private static final String ENV_SHARD_ID = "SHARD_ID";
    private static final int maxJobNameSize = 55;
    private static final int maxLabelSize = 63;
    public static final Pattern VALID_POD_NAME_REGEX =
            Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*",
                    Pattern.CASE_INSENSITIVE);
    private static final String PULSARFUNCTIONS_CONTAINER_NAME = "pulsarfunction";

    private final AppsV1Api appsClient;
    private final CoreV1Api coreClient;
    static final List<String> TOLERATIONS = Collections.unmodifiableList(
            Arrays.asList(
                    "node.kubernetes.io/not-ready",
                    "node.alpha.kubernetes.io/notReady",
                    "node.alpha.kubernetes.io/unreachable"
            )
    );
    private static final long GRPC_TIMEOUT_SECS = 5;
    private final boolean authenticationEnabled;

    // The thread that invokes the function
    @Getter
    private List<String> processArgs;
    @Getter
    private ManagedChannel[] channel;
    private InstanceControlGrpc.InstanceControlFutureStub[] stub;
    private InstanceConfig instanceConfig;
    private final String jobNamespace;
    private final String jobName;
    private final Map<String, String> customLabels;
    private final Map<String, String> functionDockerImages;
    private final String pulsarDockerImageName;
    private final String imagePullPolicy;
    private final String pulsarRootDir;
    private final String configAdminCLI;
    private final String userCodePkgUrl;
    private final String originalCodeFileName;
    private final String pulsarAdminUrl;
    private final SecretsProviderConfigurator secretsProviderConfigurator;
    private int percentMemoryPadding;
    private double cpuOverCommitRatio;
    private double memoryOverCommitRatio;
    private final Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider;
    private final AuthenticationConfig authConfig;
    private Integer grpcPort;
    private Integer metricsPort;
    private String narExtractionDirectory;
    private final Optional<KubernetesManifestCustomizer> manifestCustomizer;
    private String functionInstanceClassPath;
    private String downloadDirectory;

    KubernetesRuntime(AppsV1Api appsClient,
                      CoreV1Api coreClient,
                      String jobNamespace,
                      String jobName,
                      Map<String, String> customLabels,
                      Boolean installUserCodeDependencies,
                      String pythonDependencyRepository,
                      String pythonExtraDependencyRepository,
                      String pulsarDockerImageName,
                      Map<String, String> functionDockerImages,
                      String imagePullPolicy,
                      String pulsarRootDir,
                      InstanceConfig instanceConfig,
                      String instanceFile,
                      String extraDependenciesDir,
                      String logDirectory,
                      String configAdminCLI,
                      String userCodePkgUrl,
                      String originalCodeFileName,
                      String pulsarServiceUrl,
                      String pulsarAdminUrl,
                      String stateStorageServiceUrl,
                      AuthenticationConfig authConfig,
                      SecretsProviderConfigurator secretsProviderConfigurator,
                      Integer expectedMetricsCollectionInterval,
                      int percentMemoryPadding,
                      double cpuOverCommitRatio,
                      double memoryOverCommitRatio,
                      Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider,
                      boolean authenticationEnabled,
                      Integer grpcPort,
                      String narExtractionDirectory,
                      Optional<KubernetesManifestCustomizer> manifestCustomizer,
                      String functionInstanceClassPath,
                      String downloadDirectory) throws Exception {
        this.appsClient = appsClient;
        this.coreClient = coreClient;
        this.instanceConfig = instanceConfig;
        this.jobNamespace = jobNamespace;
        this.jobName = jobName;
        this.customLabels = customLabels;
        this.functionDockerImages = functionDockerImages;
        this.pulsarDockerImageName = pulsarDockerImageName;
        this.imagePullPolicy = imagePullPolicy;
        this.pulsarRootDir = pulsarRootDir;
        this.configAdminCLI = configAdminCLI;
        this.userCodePkgUrl = userCodePkgUrl;
        this.downloadDirectory = StringUtils.isNotEmpty(downloadDirectory) ? downloadDirectory : this.pulsarRootDir; // for backward comp
        this.originalCodeFileName = this.downloadDirectory + "/" + originalCodeFileName;
        this.pulsarAdminUrl = pulsarAdminUrl;
        this.secretsProviderConfigurator = secretsProviderConfigurator;
        this.percentMemoryPadding = percentMemoryPadding;
        this.cpuOverCommitRatio = cpuOverCommitRatio;
        this.memoryOverCommitRatio = memoryOverCommitRatio;
        this.authenticationEnabled = authenticationEnabled;
        this.manifestCustomizer = manifestCustomizer;
        this.functionInstanceClassPath = functionInstanceClassPath;
        String logConfigFile = null;
        String secretsProviderClassName = secretsProviderConfigurator.getSecretsProviderClassName(instanceConfig.getFunctionDetails());
        String secretsProviderConfig = null;
        if (secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()) != null) {
            secretsProviderConfig = new Gson().toJson(secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()));
        }
        switch (instanceConfig.getFunctionDetails().getRuntime()) {
            case JAVA:
                logConfigFile = "kubernetes_instance_log4j2.xml";
                break;
            case PYTHON:
                logConfigFile = pulsarRootDir + "/conf/functions-logging/console_logging_config.ini";
                break;
            case GO:
                break;
        }

        this.authConfig = authConfig;

        this.functionAuthDataCacheProvider = functionAuthDataCacheProvider;

        this.grpcPort = grpcPort;
        this.metricsPort = instanceConfig.hasValidMetricsPort() ? instanceConfig.getMetricsPort() : null;
        this.narExtractionDirectory = narExtractionDirectory;

        this.processArgs = new LinkedList<>();
        this.processArgs.addAll(RuntimeUtils.getArgsBeforeCmd(instanceConfig, extraDependenciesDir));

        if (instanceConfig.getFunctionDetails().getRuntime() == Function.FunctionDetails.Runtime.GO) {
            // before we run the command, make sure the go executable with correct permissions
            this.processArgs.add("chmod");
            this.processArgs.add("777");
            this.processArgs.add(this.originalCodeFileName);
            this.processArgs.add("&&");
        }

        // use exec to to launch function so that it gets launched in the foreground with the same PID as shell
        // so that when we kill the pod, the signal will get propagated to the function code
        this.processArgs.add("exec");

        this.processArgs.addAll(
                RuntimeUtils.getCmd(
                        instanceConfig,
                        instanceFile,
                        extraDependenciesDir,
                        logDirectory,
                        this.originalCodeFileName,
                        pulsarServiceUrl,
                        stateStorageServiceUrl,
                        authConfig,
                        "$" + ENV_SHARD_ID,
                        grpcPort,
                        -1L,
                        logConfigFile,
                        secretsProviderClassName,
                        secretsProviderConfig,
                        installUserCodeDependencies,
                        pythonDependencyRepository,
                        pythonExtraDependencyRepository,
                        narExtractionDirectory,
                        functionInstanceClassPath,
                        true,
                        pulsarAdminUrl));

        doChecks(instanceConfig.getFunctionDetails(), this.jobName);
    }

    /**
     * The core logic that creates a service first followed by statefulset.
     */
    @Override
    public void start() throws Exception {

        try {
            submitService();
            submitStatefulSet();

        } catch (Exception e) {
            log.error("Failed start function {}/{}/{} in Kubernetes",
                    instanceConfig.getFunctionDetails().getTenant(),
                    instanceConfig.getFunctionDetails().getNamespace(),
                    instanceConfig.getFunctionDetails().getName(), e);
            stop();
            throw e;
        }

        setupGrpcChannelIfNeeded();
    }

    @Override
    public void reinitialize() {
        setupGrpcChannelIfNeeded();
    }

    private synchronized void setupGrpcChannelIfNeeded() {
        if (channel == null || stub == null) {
            channel = new ManagedChannel[instanceConfig.getFunctionDetails().getParallelism()];
            stub = new InstanceControlGrpc.InstanceControlFutureStub[instanceConfig.getFunctionDetails().getParallelism()];

            String jobName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);
            for (int i = 0; i < instanceConfig.getFunctionDetails().getParallelism(); ++i) {
                String address = getServiceUrl(jobName, jobNamespace, i);
                channel[i] = ManagedChannelBuilder.forAddress(address, grpcPort)
                        .usePlaintext()
                        .build();
                stub[i] = InstanceControlGrpc.newFutureStub(channel[i]);
            }
        }
    }

    @Override
    public void join() throws Exception {
        // K8 functions never return
        this.wait();
    }

    @Override
    public void stop() throws Exception {
        deleteStatefulSet();
        deleteService();

        if (channel != null) {
            for (ManagedChannel cn : channel) {
                cn.shutdown();
            }
        }
        channel = null;
        stub = null;
    }

    @Override
    public Throwable getDeathException() {
        return null;
    }

    @Override
    public CompletableFuture<FunctionStatus> getFunctionStatus(int instanceId) {
        CompletableFuture<FunctionStatus> retval = new CompletableFuture<>();
        if (stub == null) {
            retval.completeExceptionally(new RuntimeException("Not alive"));
            return retval;
        }
        if (instanceId < 0 || instanceId >= stub.length) {
            retval.completeExceptionally(new RuntimeException("Invalid InstanceId"));
            return retval;
        }
        ListenableFuture<FunctionStatus> response = stub[instanceId].withDeadlineAfter(GRPC_TIMEOUT_SECS, TimeUnit.SECONDS).getFunctionStatus(Empty.newBuilder().build());
        Futures.addCallback(response, new FutureCallback<FunctionStatus>() {
            @Override
            public void onFailure(Throwable throwable) {
                FunctionStatus.Builder builder = FunctionStatus.newBuilder();
                builder.setRunning(false);
                builder.setFailureException(throwable.getMessage());
                retval.complete(builder.build());
            }

            @Override
            public void onSuccess(FunctionStatus t) {
                retval.complete(t);
            }
        }, MoreExecutors.directExecutor());
        return retval;
    }

    @Override
    public CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics() {
        CompletableFuture<InstanceCommunication.MetricsData> retval = new CompletableFuture<>();
        retval.completeExceptionally(new RuntimeException("Kubernetes Runtime doesn't support getAndReset metrics via rest"));
        return retval;
    }

    @Override
    public CompletableFuture<Void> resetMetrics() {
        CompletableFuture<Void> retval = new CompletableFuture<>();
        retval.completeExceptionally(new RuntimeException("Kubernetes Runtime doesn't support resetting metrics via rest"));
        return retval;
    }

    @Override
    public CompletableFuture<InstanceCommunication.MetricsData> getMetrics(int instanceId) {
        CompletableFuture<InstanceCommunication.MetricsData> retval = new CompletableFuture<>();
        if (stub == null) {
            retval.completeExceptionally(new RuntimeException("Not alive"));
            return retval;
        }

        if (instanceId < 0 || instanceId >= stub.length) {
            retval.completeExceptionally(new RuntimeException("Invalid InstanceId"));
            return retval;
        }

        ListenableFuture<InstanceCommunication.MetricsData> response = stub[instanceId].withDeadlineAfter(GRPC_TIMEOUT_SECS, TimeUnit.SECONDS).getMetrics(Empty.newBuilder().build());
        Futures.addCallback(response, new FutureCallback<InstanceCommunication.MetricsData>() {
            @Override
            public void onFailure(Throwable throwable) {
                InstanceCommunication.MetricsData.Builder builder = InstanceCommunication.MetricsData.newBuilder();
                retval.complete(builder.build());
            }

            @Override
            public void onSuccess(InstanceCommunication.MetricsData t) {
                retval.complete(t);
            }
        }, MoreExecutors.directExecutor());
        return retval;
    }

    @Override
    public String getPrometheusMetrics() throws IOException {
        return RuntimeUtils.getPrometheusMetrics(metricsPort);
    }

    @Override
    public boolean isAlive() {
        // No point for kubernetes just return dummy value
        return true;
    }

    private void submitService() throws Exception {
        final V1Service service = createService();
        log.info("Submitting the following service to k8 {}", coreClient.getApiClient().getJSON().serialize(service));

        String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());

        Actions.Action createService = Actions.Action.builder()
                .actionName(String.format("Submitting service for function %s", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS)
                .supplier(() -> {
                    final V1Service response;
                    try {
                        response = coreClient.createNamespacedService(jobNamespace, service, null, null, null);
                    } catch (ApiException e) {
                        // already exists
                        if (e.getCode() == HTTP_CONFLICT) {
                            log.warn("Service already present for function {}", fqfn);
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }

                    return Actions.ActionResult.builder().success(true).build();
                })
                .build();


        AtomicBoolean success = new AtomicBoolean(false);
        Actions.newBuilder()
                .addAction(createService.toBuilder()
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .run();

        if (!success.get()) {
            throw new RuntimeException(String.format("Failed to create service for function %s", fqfn));
        }
    }

    @VisibleForTesting
    V1Service createService() {
        final String jobName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);

        final V1Service service = new V1Service();

        // setup stateful set metadata
        final V1ObjectMeta objectMeta = new V1ObjectMeta();
        objectMeta.name(jobName);
        objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails()));
        // we don't technically need to set this, but it is useful for testing
        objectMeta.setNamespace(jobNamespace);
        service.metadata(objectMeta);

        // create the stateful set spec
        final V1ServiceSpec serviceSpec = new V1ServiceSpec();

        serviceSpec.clusterIP("None");

        final V1ServicePort servicePort = new V1ServicePort();
        servicePort.name("grpc").port(grpcPort).protocol("TCP");
        serviceSpec.addPortsItem(servicePort);

        serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails()));

        service.spec(serviceSpec);

        // let the customizer run but ensure it doesn't change the name so we can find it again
        final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service);
        overridden.getMetadata().name(jobName);

        return overridden;
    }

    private void submitStatefulSet() throws Exception {
        final V1StatefulSet statefulSet = createStatefulSet();
        // Configure function authentication if needed
        if (authenticationEnabled) {
            functionAuthDataCacheProvider.ifPresent(kubernetesFunctionAuthProvider -> kubernetesFunctionAuthProvider.configureAuthDataStatefulSet(
                    statefulSet, Optional.ofNullable(getFunctionAuthData(Optional.ofNullable(instanceConfig.getFunctionAuthenticationSpec())))));
        }

        log.info("Submitting the following spec to k8 {}", appsClient.getApiClient().getJSON().serialize(statefulSet));

        String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());

        Actions.Action createStatefulSet = Actions.Action.builder()
                .actionName(String.format("Submitting statefulset for function %s", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS)
                .supplier(() -> {
                    final V1StatefulSet response;
                    try {
                        response = appsClient.createNamespacedStatefulSet(jobNamespace, statefulSet, null, null, null);
                    } catch (ApiException e) {
                        // already exists
                        if (e.getCode() == HTTP_CONFLICT) {
                            log.warn("Statefulset already present for function {}", fqfn);
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }

                    return Actions.ActionResult.builder().success(true).build();
                })
                .build();


        AtomicBoolean success = new AtomicBoolean(false);
        Actions.newBuilder()
                .addAction(createStatefulSet.toBuilder()
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .run();

        if (!success.get()) {
            throw new RuntimeException(String.format("Failed to create statefulset for function %s", fqfn));
        }
    }


    public void deleteStatefulSet() throws InterruptedException {
        String statefulSetName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);
        final V1DeleteOptions options = new V1DeleteOptions();
        options.setGracePeriodSeconds(5L);
        options.setPropagationPolicy("Foreground");

        String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());
        Actions.Action deleteStatefulSet = Actions.Action.builder()
                .actionName(String.format("Deleting statefulset for function %s", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS)
                .supplier(() -> {
                    Response response;
                    try {
                        // cannot use deleteNamespacedStatefulSet because of bug in kuberenetes
                        // https://github.com/kubernetes-client/java/issues/86
                        response = appsClient.deleteNamespacedStatefulSetCall(
                                statefulSetName,
                                jobNamespace, null, null,
                                5, null, "Foreground",
                                null, null)
                                .execute();
                    } catch (ApiException e) {
                        // if already deleted
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            log.warn("Statefulset for function {} does not exist", fqfn);
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    } catch (IOException e) {
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(e.getMessage())
                                .build();
                    }

                    // if already deleted
                    if (response.code() == HTTP_NOT_FOUND) {
                        log.warn("Statefulset for function {} does not exist", fqfn);
                        return Actions.ActionResult.builder().success(true).build();
                    } else {
                        return Actions.ActionResult.builder()
                                .success(response.isSuccessful())
                                .errorMsg(response.message())
                                .build();
                    }
                })
                .build();


        Actions.Action waitForStatefulSetDeletion = Actions.Action.builder()
                .actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
                // set retry period to be about 2x the graceshutdown time
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES * 2)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS* 2)
                .supplier(() -> {
                    V1StatefulSet response;
                    try {
                        response = appsClient.readNamespacedStatefulSet(statefulSetName, jobNamespace,
                                null, null, null);
                    } catch (ApiException e) {
                        // statefulset is gone
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }
                    return Actions.ActionResult.builder()
                            .success(false)
                            .errorMsg(response.getStatus().toString())
                            .build();
                })
                .build();

        // Need to wait for all pods to die so we can cleanup subscriptions.
        Actions.Action waitForStatefulPodsToTerminate = Actions.Action.builder()
                .actionName(String.format("Waiting for pods for function %s to terminate", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES * 2)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS * 2)
                .supplier(() -> {
                    String labels = String.format("tenant=%s,namespace=%s,name=%s",
                            instanceConfig.getFunctionDetails().getTenant(),
                            instanceConfig.getFunctionDetails().getNamespace(),
                            instanceConfig.getFunctionDetails().getName());

                    V1PodList response;
                    try {
                        response = coreClient.listNamespacedPod(jobNamespace, null, null,
                                null, null, labels,
                                null, null, null, null);
                    } catch (ApiException e) {

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }

                    if (response.getItems().size() > 0) {
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(response.getItems().size() + " pods still alive.")
                                .build();
                    } else {
                        return Actions.ActionResult.builder()
                                .success(true)
                                .build();
                    }
                })
                .build();


        AtomicBoolean success = new AtomicBoolean(false);
        Actions.newBuilder()
                .addAction(deleteStatefulSet.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForStatefulSetDeletion.toBuilder()
                        .continueOn(false)
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .addAction(deleteStatefulSet.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForStatefulSetDeletion.toBuilder()
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .run();

        if (!success.get()) {
            throw new RuntimeException(String.format("Failed to delete statefulset for function %s", fqfn));
        } else {
            // wait for pods to terminate
            Actions.newBuilder()
                    .addAction(waitForStatefulPodsToTerminate)
                    .run();
        }
    }

    public void deleteService() throws InterruptedException {

        final V1DeleteOptions options = new V1DeleteOptions();
        options.setGracePeriodSeconds(0L);
        options.setPropagationPolicy("Foreground");
        String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());
        String serviceName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);

        Actions.Action deleteService = Actions.Action.builder()
                .actionName(String.format("Deleting service for function %s", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS)
                .supplier(() -> {
                    final Response response;
                    try {
                        // cannot use deleteNamespacedService because of bug in kuberenetes
                        // https://github.com/kubernetes-client/java/issues/86
                        response = coreClient.deleteNamespacedServiceCall(
                                serviceName,
                                jobNamespace, null, null,
                                0, null,
                                "Foreground", null, null).execute();
                    } catch (ApiException e) {
                        // if already deleted
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            log.warn("Service for function {} does not exist", fqfn);
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    } catch (IOException e) {
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(e.getMessage())
                                .build();
                    }

                    // if already deleted
                    if (response.code() == HTTP_NOT_FOUND) {
                        log.warn("Service for function {} does not exist", fqfn);
                        return Actions.ActionResult.builder().success(true).build();
                    } else {
                        return Actions.ActionResult.builder()
                                .success(response.isSuccessful())
                                .errorMsg(response.message())
                                .build();
                    }
                })
                .build();

        Actions.Action waitForServiceDeletion = Actions.Action.builder()
                .actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
                .numRetries(KubernetesRuntimeFactory.NUM_RETRIES)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.SLEEP_BETWEEN_RETRIES_MS)
                .supplier(() -> {
                    V1Service response;
                    try {
                        response = coreClient.readNamespacedService(serviceName, jobNamespace,
                                null, null, null);

                    } catch (ApiException e) {
                        // statefulset is gone
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            return Actions.ActionResult.builder().success(true).build();
                        }
                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }
                    return Actions.ActionResult.builder()
                            .success(false)
                            .errorMsg(response.getStatus().toString())
                            .build();
                })
                .build();

        AtomicBoolean success = new AtomicBoolean(false);
        Actions.newBuilder()
                .addAction(deleteService.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForServiceDeletion.toBuilder()
                        .continueOn(false)
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .addAction(deleteService.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForServiceDeletion.toBuilder()
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .run();

        if (!success.get()) {
            throw new RuntimeException(String.format("Failed to delete service for function %s", fqfn));
        }
    }

    protected List<String> getExecutorCommand() {
        return Arrays.asList(
                "sh",
                "-c",
                String.join(" ", getDownloadCommand(instanceConfig.getFunctionDetails(), originalCodeFileName))
                        + " && " + setShardIdEnvironmentVariableCommand()
                        + " && " + String.join(" ", processArgs)
        );
    }

    private List<String> getDownloadCommand(Function.FunctionDetails functionDetails, String userCodeFilePath) {
        if (Arrays.stream(PackageType.values()).anyMatch(type ->
            functionDetails.getPackageUrl().startsWith(type.toString()))) {
            return getPackageDownloadCommand(functionDetails.getPackageUrl(), userCodeFilePath);
        } else {
            return getDownloadCommand(functionDetails.getTenant(), functionDetails.getNamespace(),
                functionDetails.getName(), userCodeFilePath);
        }
    }

    private List<String> getDownloadCommand(String tenant, String namespace, String name, String userCodeFilePath) {

        // add auth plugin and parameters if necessary
        if (authenticationEnabled && authConfig != null) {
            if (isNotBlank(authConfig.getClientAuthenticationPlugin())
                    && isNotBlank(authConfig.getClientAuthenticationParameters())
                    && instanceConfig.getFunctionAuthenticationSpec() != null) {
                return Arrays.asList(
                        pulsarRootDir + configAdminCLI,
                        "--auth-plugin",
                        authConfig.getClientAuthenticationPlugin(),
                        "--auth-params",
                        authConfig.getClientAuthenticationParameters(),
                        "--admin-url",
                        pulsarAdminUrl,
                        "functions",
                        "download",
                        "--tenant",
                        tenant,
                        "--namespace",
                        namespace,
                        "--name",
                        name,
                        "--destination-file",
                        userCodeFilePath);
            }
        }

        return Arrays.asList(
                pulsarRootDir + configAdminCLI,
                "--admin-url",
                pulsarAdminUrl,
                "functions",
                "download",
                "--tenant",
                tenant,
                "--namespace",
                namespace,
                "--name",
                name,
                "--destination-file",
                userCodeFilePath);
    }

    private List<String> getPackageDownloadCommand(String packageName, String userCodeFilePath) {
        // add auth plugin and parameters if necessary
        if (authenticationEnabled && authConfig != null) {
            if (isNotBlank(authConfig.getClientAuthenticationPlugin())
                && isNotBlank(authConfig.getClientAuthenticationParameters())
                && instanceConfig.getFunctionAuthenticationSpec() != null) {
                return Arrays.asList(
                    pulsarRootDir + configAdminCLI,
                    "--auth-plugin",
                    authConfig.getClientAuthenticationPlugin(),
                    "--auth-params",
                    authConfig.getClientAuthenticationParameters(),
                    "--admin-url",
                    pulsarAdminUrl,
                    "packages",
                    "download",
                    packageName,
                    "--path",
                    userCodeFilePath);
            }
        }

        return Arrays.asList(
            pulsarRootDir + configAdminCLI,
            "--admin-url",
            pulsarAdminUrl,
            "packages",
            "download",
            packageName,
            "--path",
            userCodeFilePath);
    }

    private static String setShardIdEnvironmentVariableCommand() {
        return String.format("%s=${POD_NAME##*-} && echo shardId=${%s}", ENV_SHARD_ID, ENV_SHARD_ID);
    }

    @VisibleForTesting
    V1StatefulSet createStatefulSet() {
        final String jobName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);

        final V1StatefulSet statefulSet = new V1StatefulSet();

        // setup stateful set metadata
        final V1ObjectMeta objectMeta = new V1ObjectMeta();
        objectMeta.name(jobName);
        objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails()));
        // we don't technically need to set this, but it is useful for testing
        objectMeta.setNamespace(jobNamespace);
        statefulSet.metadata(objectMeta);

        // create the stateful set spec
        final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec();
        statefulSetSpec.serviceName(jobName);
        statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism());

        // Parallel pod management tells the StatefulSet controller to launch or terminate
        // all Pods in parallel, and not to wait for Pods to become Running and Ready or completely
        // terminated prior to launching or terminating another Pod.
        statefulSetSpec.setPodManagementPolicy("Parallel");

        // add selector match labels
        // so the we know which pods to manage
        final V1LabelSelector selector = new V1LabelSelector();
        selector.matchLabels(getLabels(instanceConfig.getFunctionDetails()));
        statefulSetSpec.selector(selector);

        // create a pod template
        final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec();

        // set up pod meta
        final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails()));
        templateMetaData.annotations(getPrometheusAnnotations());
        podTemplateSpec.setMetadata(templateMetaData);

        final List<String> command = getExecutorCommand();
        podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null));

        statefulSetSpec.setTemplate(podTemplateSpec);

        statefulSet.spec(statefulSetSpec);

        // let the customizer run but ensure it doesn't change the name so we can find it again
        final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet);
        overridden.getMetadata().name(jobName);

        return statefulSet;
    }

    private Map<String, String> getPrometheusAnnotations() {
        final Map<String, String> annotations = new HashMap<>();
        annotations.put("prometheus.io/scrape", "true");
        annotations.put("prometheus.io/port", String.valueOf(metricsPort));
        return annotations;
    }

    private Map<String, String> getLabels(Function.FunctionDetails functionDetails) {
        final Map<String, String> labels = new HashMap<>();
        Function.FunctionDetails.ComponentType componentType = InstanceUtils.calculateSubjectType(functionDetails);
        String component;
        switch (componentType) {
            case FUNCTION:
                component = "function";
                break;
            case SOURCE:
                component = "source";
                break;
            case SINK:
                component = "sink";
                break;
            default:
                component = "function";
                break;
        }
        labels.put("component", component);
        labels.put("namespace", toValidLabelName(functionDetails.getNamespace()));
        labels.put("tenant", toValidLabelName(functionDetails.getTenant()));
        labels.put("name", toValidLabelName(functionDetails.getName()));
        if (customLabels != null && !customLabels.isEmpty()) {
            customLabels.replaceAll((k, v) -> toValidLabelName(v));
            labels.putAll(customLabels);
        }
        return labels;
    }

    private V1PodSpec getPodSpec(List<String> instanceCommand, Function.Resources resource) {
        final V1PodSpec podSpec = new V1PodSpec();

        // set the termination period to 0 so pods can be deleted quickly
        podSpec.setTerminationGracePeriodSeconds(0L);

        // set the pod tolerations so pods are rescheduled when nodes go down
        // https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/#taint-based-evictions
        podSpec.setTolerations(getTolerations());

        List<V1Container> containers = new LinkedList<>();
        containers.add(getFunctionContainer(instanceCommand, resource));
        podSpec.containers(containers);

        // Configure secrets
        secretsProviderConfigurator.configureKubernetesRuntimeSecretsProvider(podSpec, PULSARFUNCTIONS_CONTAINER_NAME, instanceConfig.getFunctionDetails());

        return podSpec;
    }

    private List<V1Toleration> getTolerations() {
        final List<V1Toleration> tolerations = new ArrayList<>();
        TOLERATIONS.forEach(t -> {
            final V1Toleration toleration =
                    new V1Toleration()
                            .key(t)
                            .operator("Exists")
                            .effect("NoExecute")
                            .tolerationSeconds(10L);
            tolerations.add(toleration);
        });

        return tolerations;
    }

    @VisibleForTesting
    V1Container getFunctionContainer(List<String> instanceCommand, Function.Resources resource) {
        final V1Container container = new V1Container().name(PULSARFUNCTIONS_CONTAINER_NAME);

        Function.FunctionDetails.Runtime runtime = instanceConfig.getFunctionDetails().getRuntime();

        String imageName = null;
        if (functionDockerImages != null) {
            switch (runtime) {
                case JAVA:
                    if (functionDockerImages.get("JAVA") != null) {
                        imageName = functionDockerImages.get("JAVA");
                        break;
                    }
                case PYTHON:
                    if (functionDockerImages.get("PYTHON") != null) {
                        imageName = functionDockerImages.get("PYTHON");
                        break;
                    }
                case GO:
                    if (functionDockerImages.get("GO") != null) {
                        imageName = functionDockerImages.get("GO");
                        break;
                    }
                default:
                    imageName = pulsarDockerImageName;
                    break;
            }
            container.setImage(imageName);
        } else {
            container.setImage(pulsarDockerImageName);
        }

        container.setImagePullPolicy(imagePullPolicy);

        // set up the container command
        container.setCommand(instanceCommand);

        // setup the environment variables for the container
        final V1EnvVar envVarPodName = new V1EnvVar();
        envVarPodName.name("POD_NAME")
                .valueFrom(new V1EnvVarSource()
                        .fieldRef(new V1ObjectFieldSelector()
                                .fieldPath("metadata.name")));
        container.addEnvItem(envVarPodName);

        // set container resources
        final V1ResourceRequirements resourceRequirements = new V1ResourceRequirements();
        final Map<String, Quantity> resourceLimit = new HashMap<>();
        final Map<String, Quantity> resourceRequest = new HashMap<>();


        long ram = resource != null && resource.getRam() != 0 ? resource.getRam() : 1073741824;

        // add memory padding
        long padding = Math.round(ram * (percentMemoryPadding / 100.0));
        long ramWithPadding = ram + padding;
        long ramRequest =  (long) (ramWithPadding / memoryOverCommitRatio);

        // set resource limits
        double cpuLimit = resource != null && resource.getCpu() != 0 ? resource.getCpu() : 1;
        // for cpu overcommiting
        double cpuRequest = cpuLimit / cpuOverCommitRatio;

        // round cpu to 3 decimal places as it is the finest cpu precision allowed
        resourceLimit.put("cpu", Quantity.fromString(Double.toString(roundDecimal(cpuLimit, 3))));
        resourceLimit.put("memory", Quantity.fromString(Long.toString(ramWithPadding)));

        // set resource requests
        // round cpu to 3 decimal places as it is the finest cpu precision allowed
        resourceRequest.put("cpu", Quantity.fromString(Double.toString(roundDecimal(cpuRequest, 3))));
        resourceRequest.put("memory", Quantity.fromString(Long.toString(ramRequest)));

        resourceRequirements.setRequests(resourceRequest);
        resourceRequirements.setLimits(resourceLimit);
        container.setResources(resourceRequirements);

        // set container ports
        container.setPorts(getFunctionContainerPorts());

        return container;
    }

    private List<V1ContainerPort> getFunctionContainerPorts() {
        List<V1ContainerPort> ports = new ArrayList<>();
        final V1ContainerPort port = new V1ContainerPort();
        port.setName("grpc");
        port.setContainerPort(grpcPort);
        ports.add(port);
        return ports;
    }

    private List<V1ContainerPort> getPrometheusContainerPorts() {
        List<V1ContainerPort> ports = new ArrayList<>();
        final V1ContainerPort port = new V1ContainerPort();
        port.setName("prometheus");
        port.setContainerPort(metricsPort);
        ports.add(port);
        return ports;
    }

    public static String createJobName(Function.FunctionDetails functionDetails, String jobName) {
        return jobName == null ? createJobName(functionDetails.getTenant(),
                functionDetails.getNamespace(), functionDetails.getName()) : 
                	createJobName(jobName, functionDetails.getTenant(),
                        functionDetails.getNamespace(), functionDetails.getName());
    }

    private static String toValidPodName(String ori) {
        return ori.toLowerCase().replaceAll("[^a-z0-9-\\.]", "-");
    }

    private static String toValidLabelName(String ori) {
        return left(ori.toLowerCase().replaceAll("[^a-zA-Z0-9-_\\.]", "-").replaceAll("^[^a-zA-Z0-9]", "0").replaceAll("[^a-zA-Z0-9]$", "0"), maxLabelSize);
    }
    
    private static String createJobName(String jobName, String tenant, String namespace, String functionName) {
    	final String convertedJobName = toValidPodName(jobName);
        // use of customRuntimeOptions 'jobName' may cause naming collisions, 
    	// add a short hash here to avoid it
    	final String hashName = String.format("%s-%s-%s-%s", jobName, tenant, namespace, functionName);
        final String shortHash = DigestUtils.sha1Hex(hashName).toLowerCase().substring(0, 8);
        return convertedJobName + "-" + shortHash;
    }
    
    private static String createJobName(String tenant, String namespace, String functionName) {
    	final String jobNameBase = String.format("%s-%s-%s", tenant, namespace, functionName);
        final String jobName = "pf-" + jobNameBase;
        final String convertedJobName = toValidPodName(jobName);
        if (jobName.equals(convertedJobName)) {
            return jobName;
        }
        // toValidPodName may cause naming collisions, add a short hash here to avoid it
        final String shortHash = DigestUtils.sha1Hex(jobNameBase).toLowerCase().substring(0, 8);
        return convertedJobName + "-" + shortHash;
    }

    private static String getServiceUrl(String jobName, String jobNamespace, int instanceId) {
        return String.format("%s-%d.%s.%s.svc.cluster.local", jobName, instanceId, jobName, jobNamespace);
    }

    public static void doChecks(Function.FunctionDetails functionDetails, String overridenJobName) {
        final String jobName = createJobName(functionDetails, overridenJobName);
        if (!jobName.equals(jobName.toLowerCase())) {
            throw new RuntimeException("Kubernetes does not allow upper case jobNames.");
        }
        final Matcher matcher = VALID_POD_NAME_REGEX.matcher(jobName);
        if (!matcher.matches()) {
            throw new RuntimeException("Kubernetes only admits lower case and numbers. " +
            		"(jobName=" + jobName + ")");
        }
        if (jobName.length() > maxJobNameSize) {
            throw new RuntimeException("Kubernetes job name size should be less than " + maxJobNameSize);
        }
    }
}