Merge branch 'master' into pip-320-metric-scaffolding

docker/pulsar/scripts/apply-config-from-env-with-prefix.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python3
+#!/usr/bin/env bash
 #
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
@@ -32,83 +32,8 @@
 # update if they exist and ignored if they don't.
 ############################################################
 
-import os
-import sys
-
-if len(sys.argv) < 3:
-    print('Usage: %s <PREFIX> <FILE> [<FILE>...]' % (sys.argv[0]))
-    sys.exit(1)
-
-# Always apply env config to env scripts as well
-prefix = sys.argv[1]
-conf_files = sys.argv[2:]
-
-PF_ENV_DEBUG = (os.environ.get('PF_ENV_DEBUG','0') == '1')
-
-for conf_filename in conf_files:
-    lines = []  # List of config file lines
-    keys = {} # Map a key to its line number in the file
-
-    # Load conf file
-    for line in open(conf_filename):
-        lines.append(line)
-        line = line.strip()
-        if not line or line.startswith('#'):
-            continue
-
-        try:
-            k,v = line.split('=', 1)
-            keys[k] = len(lines) - 1
-        except:
-            if PF_ENV_DEBUG:
-                print("[%s] skip Processing %s" % (conf_filename, line))
-
-    # Update values from Env
-    for k in sorted(os.environ.keys()):
-        v = os.environ[k].strip()
-
-        # Hide the value in logs if is password.
-        if "password" in k.lower():
-            displayValue = "********"
-        else:
-            displayValue = v
-
-        if k.startswith(prefix):
-            k = k[len(prefix):]
-        if k in keys:
-            print('[%s] Applying config %s = %s' % (conf_filename, k, displayValue))
-            idx = keys[k]
-            lines[idx] = '%s=%s\n' % (k, v)
-
-
-    # Ensure we have a new-line at the end of the file, to avoid issue
-    # when appending more lines to the config
-    lines.append('\n')
-
-    # Add new keys from Env
-    for k in sorted(os.environ.keys()):
-        v = os.environ[k]
-        if not k.startswith(prefix):
-            continue
-
-        # Hide the value in logs if is password.
-        if "password" in k.lower():
-            displayValue = "********"
-        else:
-            displayValue = v
-
-        k = k[len(prefix):]
-        if k not in keys:
-            print('[%s] Adding config %s = %s' % (conf_filename, k, displayValue))
-            lines.append('%s=%s\n' % (k, v))
-        else:
-            print('[%s] Updating config %s = %s' % (conf_filename, k, displayValue))
-            lines[keys[k]] = '%s=%s\n' % (k, v)
-
-
-    # Store back the updated config in the same file
-    f = open(conf_filename, 'w')
-    for line in lines:
-        f.write(line)
-    f.close()
+# DEPRECATED: Use "apply-config-from-env.py --prefix MY_PREFIX_ conf_file" instead
 
+# this is not a python script, but a bash script. Call apply-config-from-env.py with the prefix argument
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)"
+"${SCRIPT_DIR}/apply-config-from-env.py" --prefix "$1" "${@:2}"

docker/pulsar/scripts/apply-config-from-env.py

@@ -25,18 +25,29 @@
 ## ./apply-config-from-env file.conf
 ##
 
-import os, sys
-
-if len(sys.argv) < 2:
-    print('Usage: %s' % (sys.argv[0]))
+import os, sys, argparse
+
+parser = argparse.ArgumentParser(description='Pulsar configuration file customizer based on environment variables')
+parser.add_argument('--prefix', default='PULSAR_PREFIX_', help='Prefix for environment variables, default is PULSAR_PREFIX_')
+parser.add_argument('conf_files', nargs='*', help='Configuration files')
+args = parser.parse_args()
+if not args.conf_files:
+    parser.print_help()
     sys.exit(1)
 
-# Always apply env config to env scripts as well
-conf_files = sys.argv[1:]
+env_prefix = args.prefix
+conf_files = args.conf_files
 
-PF_ENV_PREFIX = 'PULSAR_PREFIX_'
 PF_ENV_DEBUG = (os.environ.get('PF_ENV_DEBUG','0') == '1')
 
+# List of keys where the value should not be displayed in logs
+sensitive_keys = ["brokerClientAuthenticationParameters", "bookkeeperClientAuthenticationParameters", "tokenSecretKey"]
+
+def sanitize_display_value(k, v):
+    if "password" in k.lower() or k in sensitive_keys or (k == "tokenSecretKey" and v.startswith("data:")):
+        return "********"
+    return v
+
 for conf_filename in conf_files:
     lines = []  # List of config file lines
     keys = {} # Map a key to its line number in the file
@@ -47,7 +58,6 @@
         line = line.strip()
         if not line:
             continue
-
         try:
             k,v = line.split('=', 1)
             if k.startswith('#'):
@@ -61,48 +71,35 @@
     for k in sorted(os.environ.keys()):
         v = os.environ[k].strip()
 
-        # Hide the value in logs if is password.
-        if "password" in k.lower():
-            displayValue = "********"
-        else:
-            displayValue = v
-
-        if k.startswith(PF_ENV_PREFIX):
-            k = k[len(PF_ENV_PREFIX):]
         if k in keys:
+            displayValue = sanitize_display_value(k, v)
             print('[%s] Applying config %s = %s' % (conf_filename, k, displayValue))
             idx = keys[k]
             lines[idx] = '%s=%s\n' % (k, v)
 
-
     # Ensure we have a new-line at the end of the file, to avoid issue
     # when appending more lines to the config
     lines.append('\n')
-    
-    # Add new keys from Env    
+
+    # Add new keys from Env
     for k in sorted(os.environ.keys()):
-        v = os.environ[k]
-        if not k.startswith(PF_ENV_PREFIX):
+        if not k.startswith(env_prefix):
             continue
 
-        # Hide the value in logs if is password.
-        if "password" in k.lower():
-            displayValue = "********"
-        else:
-            displayValue = v
+        v = os.environ[k].strip()
+        k = k[len(env_prefix):]
+
+        displayValue = sanitize_display_value(k, v)
 
-        k = k[len(PF_ENV_PREFIX):]
         if k not in keys:
             print('[%s] Adding config %s = %s' % (conf_filename, k, displayValue))
             lines.append('%s=%s\n' % (k, v))
         else:
             print('[%s] Updating config %s = %s' % (conf_filename, k, displayValue))
             lines[keys[k]] = '%s=%s\n' % (k, v)
 
-
     # Store back the updated config in the same file
     f = open(conf_filename, 'w')
     for line in lines:
         f.write(line)
-    f.close()
-
+    f.close()

tests/docker-images/latest-version-image/Dockerfile

@@ -50,7 +50,7 @@ COPY conf/supervisord.conf /etc/supervisord.conf
 COPY conf/global-zk.conf conf/local-zk.conf conf/bookie.conf conf/broker.conf conf/functions_worker.conf \
      conf/proxy.conf conf/websocket.conf /etc/supervisord/conf.d/
 
-COPY scripts/init-cluster.sh scripts/run-global-zk.sh scripts/run-local-zk.sh \
+COPY scripts/run-global-zk.sh scripts/run-local-zk.sh \
      scripts/run-bookie.sh scripts/run-broker.sh scripts/run-functions-worker.sh scripts/run-proxy.sh \
      scripts/run-standalone.sh scripts/run-websocket.sh \
      /pulsar/bin/

tests/docker-images/latest-version-image/scripts/run-bookie.sh

@@ -29,6 +29,5 @@ if [ -z "$NO_AUTOSTART" ]; then
     sed -i 's/autostart=.*/autostart=true/' /etc/supervisord/conf.d/bookie.conf
 fi
 
-bin/watch-znode.py -z $zkServers -p /initialized-$clusterName -w
 exec /usr/bin/supervisord -c /etc/supervisord.conf
 

tests/docker-images/latest-version-image/scripts/run-broker.sh

@@ -25,6 +25,5 @@ if [ -z "$NO_AUTOSTART" ]; then
     sed -i 's/autostart=.*/autostart=true/' /etc/supervisord/conf.d/broker.conf
 fi
 
-bin/watch-znode.py -z $zookeeperServers -p /initialized-$clusterName -w
 exec /usr/bin/supervisord -c /etc/supervisord.conf
 

tests/docker-images/latest-version-image/scripts/run-functions-worker.sh

@@ -26,6 +26,5 @@ if [ -z "$NO_AUTOSTART" ]; then
     sed -i 's/autostart=.*/autostart=true/' /etc/supervisord/conf.d/functions_worker.conf
 fi
 
-bin/watch-znode.py -z $zookeeperServers -p /initialized-$clusterName -w
 exec /usr/bin/supervisord -c /etc/supervisord.conf
 

tests/docker-images/latest-version-image/scripts/run-proxy.sh

@@ -25,5 +25,4 @@ if [ -z "$NO_AUTOSTART" ]; then
     sed -i 's/autostart=.*/autostart=true/' /etc/supervisord/conf.d/proxy.conf
 fi
 
-bin/watch-znode.py -z $zookeeperServers -p /initialized-$clusterName -w
 exec /usr/bin/supervisord -c /etc/supervisord.conf

tests/docker-images/latest-version-image/scripts/run-websocket.sh

@@ -25,5 +25,4 @@ if [ -z "$NO_AUTOSTART" ]; then
     sed -i 's/autostart=.*/autostart=true/' /etc/supervisord/conf.d/websocket.conf
 fi
 
-bin/watch-znode.py -z $zookeeperServers -p /initialized-$clusterName -w
 exec /usr/bin/supervisord -c /etc/supervisord.conf

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarInitMetadataContainer.java

@@ -0,0 +1,76 @@
+/*
+ * 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.tests.integration.containers;
+
+import java.io.IOException;
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+
+/**
+ * Initialize the Pulsar metadata
+ */
+@Slf4j
+public class PulsarInitMetadataContainer extends GenericContainer<PulsarInitMetadataContainer> {
+
+    public static final String NAME = "init-metadata";
+
+    private final String clusterName;
+    private final String metadataStoreUrl;
+    private final String configurationMetadataStoreUrl;
+    private final String brokerHostname;
+
+    public PulsarInitMetadataContainer(Network network,
+                                       String clusterName,
+                                       String metadataStoreUrl,
+                                       String configurationMetadataStoreUrl,
+                                       String brokerHostname) {
+        this.clusterName = clusterName;
+        this.metadataStoreUrl = metadataStoreUrl;
+        this.configurationMetadataStoreUrl = configurationMetadataStoreUrl;
+        this.brokerHostname = brokerHostname;
+        setDockerImageName(PulsarContainer.DEFAULT_IMAGE_NAME);
+        withNetwork(network);
+
+        setCommand("sleep 1000000");
+    }
+
+
+    public void initialize() throws Exception {
+        start();
+        ExecResult res = this.execInContainer(
+                "/pulsar/bin/pulsar", "initialize-cluster-metadata",
+                "--cluster", clusterName,
+                "--metadata-store", metadataStoreUrl,
+                "--configuration-metadata-store", configurationMetadataStoreUrl,
+                "--web-service-url", "http://" + brokerHostname + ":8080/",
+                "--broker-service-url", "pulsar://" + brokerHostname + ":6650/"
+        );
+
+        if (res.getExitCode() == 0) {
+            log.info("Successfully initialized cluster");
+        } else {
+            log.warn("Failed to initialize Pulsar cluster. exit code: " + res.getExitCode());
+            log.warn("STDOUT: " + res.getStdout());
+            log.warn("STDERR: " + res.getStderr());
+            throw new IOException("Failed to initialized Pulsar Cluster");
+        }
+    }
+}