Properties getTemplateBindings()

in zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sRemoteInterpreterProcess.java [297:399]
82 lines of code
15 McCabe index (conditional complexity)

  Properties getTemplateBindings(String userName) {
    Properties k8sProperties = new Properties();

    // k8s template properties
    k8sProperties.put("zeppelin.k8s.interpreter.user", String.valueOf(userName).trim());
    k8sProperties.put("zeppelin.k8s.interpreter.namespace", getInterpreterNamespace());
    k8sProperties.put("zeppelin.k8s.interpreter.pod.name", getPodName());
    k8sProperties.put("zeppelin.k8s.interpreter.serviceAccount", getServiceAccount());
    k8sProperties.put("zeppelin.k8s.interpreter.container.name", interpreterGroupName.toLowerCase());
    k8sProperties.put("zeppelin.k8s.interpreter.container.image", containerImage);
    k8sProperties.put("zeppelin.k8s.interpreter.group.id", getInterpreterGroupId());
    k8sProperties.put("zeppelin.k8s.interpreter.group.name", interpreterGroupName);
    k8sProperties.put("zeppelin.k8s.interpreter.setting.name", getInterpreterSettingName());
    k8sProperties.put("zeppelin.k8s.interpreter.localRepo", getLocalRepoDir());
    k8sProperties.put("zeppelin.k8s.interpreter.rpc.portRange", getInterpreterPortRange());
    k8sProperties.put("zeppelin.k8s.server.rpc.service", intpEventServerHost);
    k8sProperties.put("zeppelin.k8s.server.rpc.portRange", intpEventServerPort);

    String serverNamespace = K8sUtils.getCurrentK8sNamespace();
    String interpreterNamespace = getInterpreterNamespace();
    //Set the owner reference (zeppelin-server pod) for garbage collection when zeppelin server and the zeppelin interpreter is in the same namespace (Kubernetes cannot specify an owner in different namespace).
    if (ownerUID() != null && ownerName() != null && StringUtils.equals(serverNamespace, interpreterNamespace)) {
      k8sProperties.put("zeppelin.k8s.server.uid", ownerUID());
      k8sProperties.put("zeppelin.k8s.server.pod.name", ownerName());
    }

    Map<String, String> k8sEnv = new HashMap<>(getEnv());
    // environment variables
    k8sEnv.put(ENV_SERVICE_DOMAIN, getEnv().getOrDefault(ENV_SERVICE_DOMAIN, System.getenv(ENV_SERVICE_DOMAIN) == null ? "local.zeppelin-project.org" : System.getenv(ENV_SERVICE_DOMAIN)));
    k8sEnv.put(ENV_ZEPPELIN_HOME, getEnv().getOrDefault(ENV_ZEPPELIN_HOME, System.getenv(ENV_ZEPPELIN_HOME)));

    if (isSpark()) {
      int webUiPort = 4040;
      k8sProperties.put("zeppelin.k8s.spark.container.image", sparkImage);

      // There is already initial value following --driver-java-options added in interpreter.sh
      // so we need to pass spark.driver.defaultJavaOptions and spark.driver.extraJavaOptions
      // as SPARK_DRIVER_EXTRAJAVAOPTIONS_CONF env variable to build spark-submit command correctly.
      StringJoiner driverExtraJavaOpts = new StringJoiner(" ");
      if (properties.containsKey(SPARK_DRIVER_DEFAULTJAVAOPTS)) {
        driverExtraJavaOpts.add((String) properties.remove(SPARK_DRIVER_DEFAULTJAVAOPTS));
      }
      if (properties.containsKey(SPARK_DRIVER_EXTRAJAVAOPTS)) {
        driverExtraJavaOpts.add((String) properties.remove(SPARK_DRIVER_EXTRAJAVAOPTS));
      }
      if (driverExtraJavaOpts.length() > 0) {
        k8sEnv.put("SPARK_DRIVER_EXTRAJAVAOPTIONS_CONF", driverExtraJavaOpts.toString());
      }

      if (isSparkOnKubernetes(properties)) {
        addSparkK8sProperties();
        k8sEnv.put("ZEPPELIN_SPARK_CONF", prepareZeppelinSparkConf(userName));
      }
      k8sEnv.put("SPARK_HOME", getEnv().getOrDefault("SPARK_HOME", "/spark"));

      // configure interpreter property "zeppelin.spark.uiWebUrl" if not defined, to enable spark ui through reverse proxy
      String webUrl = (String) properties.get("zeppelin.spark.uiWebUrl");
      if (StringUtils.isBlank(webUrl)) {
        webUrl = "//{{PORT}}-{{SERVICE_NAME}}.{{SERVICE_DOMAIN}}";
      }
      properties.put("zeppelin.spark.uiWebUrl",
          sparkUiWebUrlFromTemplate(
              webUrl,
              webUiPort,
              getPodName(),
              k8sEnv.get(ENV_SERVICE_DOMAIN)
          ));

      // configure interpreter property "zeppelin.k8s.spark.ingress.host" if not defined, to enable spark ui through ingress
      String ingressHost = (String) properties.get("zeppelin.k8s.spark.ingress.host");
      if (StringUtils.isBlank(ingressHost)) {
        ingressHost = "{{PORT}}-{{SERVICE_NAME}}.{{SERVICE_DOMAIN}}";
      }
      properties.put("zeppelin.k8s.spark.ingress.host",
          sparkUiWebUrlFromTemplate(
              ingressHost,
              webUiPort,
              getPodName(),
              k8sEnv.get(ENV_SERVICE_DOMAIN)
          ));

      // Resources of Interpreter Pod
      if (properties.containsKey(SPARK_DRIVER_MEMORY)) {
        String memory;
        if (properties.containsKey(SPARK_DRIVER_MEMORY_OVERHEAD)) {
          memory = K8sUtils.calculateSparkMemory(properties.getProperty(SPARK_DRIVER_MEMORY),
                                                 properties.getProperty(SPARK_DRIVER_MEMORY_OVERHEAD));
        } else {
          memory = K8sUtils.calculateMemoryWithDefaultOverhead(properties.getProperty(SPARK_DRIVER_MEMORY));
        }
        k8sProperties.put("zeppelin.k8s.interpreter.memory", memory);
      }
      if (properties.containsKey(SPARK_DRIVER_CORES)) {
        k8sProperties.put("zeppelin.k8s.interpreter.cores", properties.getProperty(SPARK_DRIVER_CORES));
      }
    }

    k8sProperties.put("zeppelin.k8s.envs", k8sEnv);

    // interpreter properties overrides the values
    k8sProperties.putAll(Maps.fromProperties(properties));
    return k8sProperties;
  }