pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [pulsar] branch master updated: [conf] Add annotations for documenting function worker configuration settings (#3117)
Date Thu, 20 Dec 2018 16:19:01 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 8c3eb34   [conf] Add annotations for documenting function worker configuration settings
(#3117)
8c3eb34 is described below

commit 8c3eb3450a64af0014c77b815635c19ed8f84cc0
Author: Sijie Guo <guosijie@gmail.com>
AuthorDate: Fri Dec 21 00:18:55 2018 +0800

     [conf] Add annotations for documenting function worker configuration settings (#3117)
    
    *Motivation*
    
    This change is adding annotations to function worker configuration for generating function
worker configuration file.
---
 .../pulsar/functions/worker/WorkerConfig.java      | 287 +++++++++++++++++++--
 1 file changed, 268 insertions(+), 19 deletions(-)

diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index d49722f..d37d93a 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -33,6 +33,8 @@ import java.util.Properties;
 import java.util.Set;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.common.configuration.Category;
+import org.apache.pulsar.common.configuration.FieldContext;
 import org.apache.pulsar.common.configuration.PulsarConfiguration;
 
 import lombok.Data;
@@ -53,54 +55,213 @@ public class WorkerConfig implements Serializable, PulsarConfiguration
{
 
     private static final long serialVersionUID = 1L;
 
+    @Category
+    private static final String CATEGORY_WORKER = "Worker Settings";
+    @Category
+    private static final String CATEGORY_FUNC_PKG = "Function Package Management";
+    @Category
+    private static final String CATEGORY_FUNC_METADATA_MNG = "Function Metadata Management";
+    @Category
+    private static final String CATEGORY_FUNC_RUNTIME_MNG = "Function Runtime Management";
+    @Category
+    private static final String CATEGORY_SECURITY = "Common Security Settings (applied for
both worker and client)";
+    @Category
+    private static final String CATEGORY_WORKER_SECURITY = "Worker Security Settings";
+    @Category
+    private static final String CATEGORY_CLIENT_SECURITY = "Security settings for clients
talking to brokers";
+    @Category
+    private static final String CATEGORY_STATE = "State Management";
+    @Category
+    private static final String CATEGORY_CONNECTORS = "Connectors";
+
+    @FieldContext(
+        category = CATEGORY_WORKER,
+        doc = "Id to identify a worker instance"
+    )
     private String workerId;
+    @FieldContext(
+        category = CATEGORY_WORKER,
+        doc = "Hostname of the worker instance"
+    )
     private String workerHostname;
+    @FieldContext(
+        category = CATEGORY_WORKER,
+        doc = "The port for serving worker http requests"
+    )
     private Integer workerPort;
+    @FieldContext(
+        category = CATEGORY_WORKER,
+        doc = "The port for serving worker https requests"
+    )
     private Integer workerPortTls;
+    @FieldContext(
+        category = CATEGORY_CONNECTORS,
+        doc = "The path to the location to locate builtin connectors"
+    )
     private String connectorsDirectory = "./connectors";
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar topic used for storing function metadata"
+    )
     private String functionMetadataTopicName;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The web service url for function workers"
+    )
     private String functionWebServiceUrl;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulser binary service url that function metadata manager talks to"
+    )
     private String pulsarServiceUrl;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar web service url that function metadata manager talks to"
+    )
     private String pulsarWebServiceUrl;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar topic used for cluster coordination"
+    )
     private String clusterCoordinationTopicName;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar namespace for storing metadata topics"
+    )
     private String pulsarFunctionsNamespace;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar cluster name. Used for creating pulsar namespace during worker
initialization"
+    )
     private String pulsarFunctionsCluster;
+    @FieldContext(
+        category = CATEGORY_FUNC_PKG,
+        doc = "The number of replicas for storing functions"
+    )
     private int numFunctionPackageReplicas;
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "The directory to download functions by runtime manager"
+    )
     private String downloadDirectory;
+    @FieldContext(
+        category = CATEGORY_STATE,
+        doc = "The service url of state storage"
+    )
     private String stateStorageServiceUrl;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The pulsar topic used for storing function assignment informations"
+    )
     private String functionAssignmentTopicName;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The scheduler class used by assigning functions to workers"
+    )
     private String schedulerClassName;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The frequency of failure checks, in milliseconds"
+    )
     private long failureCheckFreqMs;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The reschedule timeout of function assignment, in milliseconds"
+    )
     private long rescheduleTimeoutMs;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The max number of retries for initial broker reconnects when function metadata
manager"
+            + " tries to create producer on metadata topics"
+    )
     private int initialBrokerReconnectMaxRetries;
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "The max number of retries for writing assignment to assignment topic"
+    )
     private int assignmentWriteMaxRetries;
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "The frequency of instance liveness check, in milliseconds"
+    )
     private long instanceLivenessCheckFreqMs;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "The authentication plugin used by function workers to talk to brokers"
+    )
     private String clientAuthenticationPlugin;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "The parameters of the authentication plugin used by function workers to talk
to brokers"
+    )
     private String clientAuthenticationParameters;
-    // Frequency how often worker performs compaction on function-topics
+    @FieldContext(
+        category = CATEGORY_FUNC_METADATA_MNG,
+        doc = "Frequency how often worker performs compaction on function-topics, in seconds"
+    )
     private long topicCompactionFrequencySec = 30 * 60; // 30 minutes
     /***** --- TLS --- ****/
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Enable TLS"
+    )
     @Deprecated
     private boolean tlsEnabled = false;
-    // Path for the TLS certificate file
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Path for the TLS certificate file"
+    )
     private String tlsCertificateFilePath;
-    // Path for the TLS private key file
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Path for the TLS private key file"
+    )
     private String tlsKeyFilePath;
-    // Path for the trusted TLS certificate file
+    @FieldContext(
+        category = CATEGORY_SECURITY,
+        doc = "Path for the trusted TLS certificate file"
+    )
     private String tlsTrustCertsFilePath = "";
-    // Accept untrusted TLS certificate from client
+    @FieldContext(
+        category = CATEGORY_SECURITY,
+        doc = "Accept untrusted TLS certificate from client"
+    )
     private boolean tlsAllowInsecureConnection = false;
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Require trusted client cert on connect"
+    )
     private boolean tlsRequireTrustedClientCertOnConnect = false;
+    @FieldContext(
+        category = CATEGORY_CLIENT_SECURITY,
+        doc = "Whether to enable TLS when clients connect to broker"
+    )
     // TLS for Functions -> Broker
     private boolean useTls = false;
+    @FieldContext(
+        category = CATEGORY_SECURITY,
+        doc = "Whether to enable hostname verification on TLS connections"
+    )
     private boolean tlsHostnameVerificationEnable = false;
-    // Enforce authentication
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Enforce authentication"
+    )
     private boolean authenticationEnabled = false;
-    // Autentication provider name list, which is a list of class names
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Autentication provider name list, which is a list of class names"
+    )
     private Set<String> authenticationProviders = Sets.newTreeSet();
-    // Enforce authorization on accessing functions admin-api
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Enforce authorization on accessing functions admin-api"
+    )
     private boolean authorizationEnabled = false;
-    // Role names that are treated as "super-user", meaning they will be able to access any
admin-api
+    @FieldContext(
+        category = CATEGORY_WORKER_SECURITY,
+        doc = "Role names that are treated as `super-user`, meaning they will be able to
access any admin-api"
+    )
     private Set<String> superUserRoles = Sets.newTreeSet();
     
     private Properties properties = new Properties();
@@ -116,8 +277,15 @@ public class WorkerConfig implements Serializable, PulsarConfiguration
{
     @EqualsAndHashCode
     @ToString
     public static class ThreadContainerFactory {
+        @FieldContext(
+            doc = "The name of thread group running function threads"
+        )
         private String threadGroupName;
     }
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "Thread based runtime settings"
+    )
     private ThreadContainerFactory threadContainerFactory;
 
     @Data
@@ -126,12 +294,29 @@ public class WorkerConfig implements Serializable, PulsarConfiguration
{
     @EqualsAndHashCode
     @ToString
     public static class ProcessContainerFactory {
+        @FieldContext(
+            doc = "The path to the java instance. Change the jar location only when you put"
+                + " the java instance jar in a different location"
+        )
         private String javaInstanceJarLocation;
+        @FieldContext(
+            doc = "The path to the python instance. Change the python instance location only"
+                + " when you put the python instance in a different location"
+        )
         private String pythonInstanceLocation;
+        @FieldContext(
+            doc = "The path to the log directory"
+        )
         private String logDirectory;
-        // the directory for dropping extra function dependencies
+        @FieldContext(
+            doc = "the directory for dropping extra function dependencies"
+        )
         private String extraFunctionDependenciesDir;
     }
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "Process based runtime settings"
+    )
     private ProcessContainerFactory processContainerFactory;
 
     @Data
@@ -140,33 +325,97 @@ public class WorkerConfig implements Serializable, PulsarConfiguration
{
     @EqualsAndHashCode
     @ToString
     public static class KubernetesContainerFactory {
+        @FieldContext(
+            doc = "Uri to kubernetes cluster, leave it to empty and it will use the kubernetes
settings in"
+                + " function worker machine"
+        )
         private String k8Uri;
+        @FieldContext(
+            doc = "The Kubernetes namespace to run the function instances. It is `default`,"
+                + " if this setting is left to be empty"
+        )
         private String jobNamespace;
+        @FieldContext(
+            doc = "The docker image used to run function instance. By default it is `apachepulsar/pulsar`"
+        )
         private String pulsarDockerImageName;
+        @FieldContext(
+            doc = "The root directory of pulsar home directory in the pulsar docker image
specified"
+                + " `pulsarDockerImageName`. By default it is under `/pulsar`. If you are
using your own"
+                + " customized image in `pulsarDockerImageName`, you need to set this setting
accordingly"
+        )
         private String pulsarRootDir;
+        @FieldContext(
+            doc = "This setting only takes effects if `k8Uri` is set to null. If your function
worker is"
+                + " also running as a k8s pod, set this to `true` is let function worker
to submit functions to"
+                + " the same k8s cluster as function worker is running. Set this to `false`
if your function worker"
+                + " is not running as a k8s pod"
+        )
         private Boolean submittingInsidePod;
+        @FieldContext(
+            doc = "The pulsar service url that pulsar functions should use to connect to
pulsar."
+                + " If it is not set, it will use the pulsar service url configured in function
worker."
+        )
         private String pulsarServiceUrl;
+        @FieldContext(
+            doc = "The pulsar admin url that pulsar functions should use to connect to pulsar."
+                + " If it is not set, it will use the pulsar admin url configured in function
worker."
+        )
         private String pulsarAdminUrl;
+        @FieldContext(
+            doc = "The flag indicates to install user code dependencies. (applied to python
package)"
+        )
         private Boolean installUserCodeDependencies;
+        @FieldContext(
+            doc = "The repository that pulsar functions use to download python dependencies"
+        )
         private String pythonDependencyRepository;
+        @FieldContext(
+            doc = "The repository that pulsar functions use to download extra python dependencies"
+        )
         private String pythonExtraDependencyRepository;
-        // the directory for dropping extra function dependencies.
-        // If it is not absolute path, it is relative to `pulsarRootDir`
+
+        @FieldContext(
+            doc = "the directory for dropping extra function dependencies. "
+                + "If it is not absolute path, it is relative to `pulsarRootDir`"
+        )
         private String extraFunctionDependenciesDir;
+        @FieldContext(
+            doc = "The custom labels that function worker uses to select the nodes for pods"
+        )
         private Map<String, String> customLabels;
+
+        @FieldContext(
+            doc = "The expected metrics collection interval, in seconds"
+        )
         private Integer expectedMetricsCollectionInterval = 30;
-        // Kubernetes Runtime will periodically checkback on
-        // this configMap if defined and if there are any changes
-        // to the kubernetes specific stuff, we apply those changes
+        @FieldContext(
+            doc = "Kubernetes Runtime will periodically checkback on"
+                + " this configMap if defined and if there are any changes"
+                + " to the kubernetes specific stuff, we apply those changes"
+        )
         private String changeConfigMap;
+        @FieldContext(
+            doc = "The namespace for storing change config map"
+        )
         private String changeConfigMapNamespace;
     }
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "Kubernetes based runtime settings"
+    )
     private KubernetesContainerFactory kubernetesContainerFactory;
 
-    // The classname of the secrets provider configurator.
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "The classname of the secrets provider configurator."
+    )
     private String secretsProviderConfiguratorClassName;
-    // Any config the secret provider configurator might need. This is passed on
-    // to the init method of the secretproviderconfigurator
+    @FieldContext(
+        category = CATEGORY_FUNC_RUNTIME_MNG,
+        doc = "Any config the secret provider configurator might need. \n\nThis is passed
on"
+            + " to the init method of the secretproviderconfigurator"
+    )
     private Map<String, String> secretsProviderConfiguratorConfig;
 
     public String getFunctionMetadataTopic() {
@@ -216,4 +465,4 @@ public class WorkerConfig implements Serializable, PulsarConfiguration
{
     public void setProperties(Properties properties) {
         this.properties = properties;
     }
-}
\ No newline at end of file
+}


Mime
View raw message