| /* |
| * 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.worker; |
| |
| import static org.apache.commons.lang3.StringUtils.isBlank; |
| import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.net.InetAddress; |
| import java.net.UnknownHostException; |
| import java.nio.file.Files; |
| import java.nio.file.Paths; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.TreeSet; |
| import lombok.AccessLevel; |
| import lombok.Data; |
| import lombok.Getter; |
| import lombok.ToString; |
| import lombok.experimental.Accessors; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider; |
| import org.apache.pulsar.common.configuration.Category; |
| import org.apache.pulsar.common.configuration.FieldContext; |
| import org.apache.pulsar.common.configuration.PulsarConfiguration; |
| import org.apache.pulsar.common.functions.Resources; |
| import org.apache.pulsar.common.nar.NarClassLoader; |
| import org.apache.pulsar.common.sasl.SaslConstants; |
| import org.apache.pulsar.functions.auth.KubernetesSecretsTokenAuthProvider; |
| import org.apache.pulsar.functions.instance.state.BKStateStoreProviderImpl; |
| import org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory; |
| import org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactoryConfig; |
| import org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactoryConfig; |
| import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig; |
| |
| @Data |
| @Accessors(chain = true) |
| @JsonIgnoreProperties(ignoreUnknown = true) |
| 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_FUNC_SCHEDULE_MNG = "Function Scheduling Management"; |
| @Category |
| private static final String CATEGORY_SECURITY = "Common Security Settings (applied for both worker and client)"; |
| @Category |
| private static final String CATEGORY_KEYSTORE_TLS = "KeyStoreTLS"; |
| @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"; |
| @Category |
| private static final String CATEGORY_FUNCTIONS = "Functions"; |
| |
| @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. Set to null to disable serving on the http port." |
| ) |
| private Integer workerPort; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "The port for serving worker https requests" |
| ) |
| private Integer workerPortTls; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Whether the '/metrics' endpoint requires authentication. Defaults to true." |
| + "'authenticationEnabled' must also be set for this to take effect." |
| ) |
| private boolean authenticateMetricsEndpoint = true; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Whether the '/metrics' endpoint should return default prometheus metrics. Defaults to false." |
| ) |
| private boolean includeStandardPrometheusMetrics = false; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Classname of Pluggable JVM GC metrics logger that can log GC specific metrics") |
| private String jvmGCMetricsLoggerClassName; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Number of threads to use for HTTP requests processing" |
| ) |
| private int numHttpServerThreads = 8; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Enable the enforcement of limits on the incoming HTTP requests" |
| ) |
| private boolean httpRequestsLimitEnabled = false; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Max HTTP requests per seconds allowed. The excess " |
| + "of requests will be rejected with HTTP code 429 (Too many requests)" |
| ) |
| private double httpRequestsMaxPerSecond = 100.0; |
| |
| @FieldContext(category = CATEGORY_WORKER, doc = "Max concurrent web requests") |
| private int maxConcurrentHttpRequests = 1024; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Capacity for thread pool queue in the HTTP server" |
| + " Default is set to 8192." |
| ) |
| private int httpServerThreadPoolQueueSize = 8192; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Capacity for accept queue in the HTTP server" |
| + " Default is set to 8192." |
| ) |
| private int httpServerAcceptQueueSize = 8192; |
| |
| @FieldContext(category = CATEGORY_WORKER, doc = "Maximum number of inbound http connections. " |
| + "(0 to disable limiting)") |
| private int maxHttpServerConnections = 2048; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| required = false, |
| deprecated = true, |
| doc = "Configuration store connection string (as a comma-separated list). Deprecated in favor of " |
| + "`configurationMetadataStoreUrl`" |
| ) |
| @Deprecated |
| private String configurationStoreServers; |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| required = false, |
| doc = "Configuration store connection string (as a comma-separated list)" |
| ) |
| private String configurationMetadataStoreUrl; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Metadata store session timeout in milliseconds." |
| ) |
| private long metadataStoreSessionTimeoutMillis = 30_000; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Metadata store operation timeout in seconds." |
| ) |
| private int metadataStoreOperationTimeoutSeconds = 30; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Metadata store cache expiry time in seconds." |
| ) |
| private int metadataStoreCacheExpirySeconds = 300; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Is metadata store read-only operations." |
| ) |
| private boolean metadataStoreAllowReadOnlyOperations; |
| |
| @Deprecated |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| deprecated = true, |
| doc = "ZooKeeper session timeout in milliseconds. " |
| + "@deprecated - Use metadataStoreSessionTimeoutMillis instead." |
| ) |
| private long zooKeeperSessionTimeoutMillis = -1; |
| |
| @Deprecated |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| deprecated = true, |
| doc = "ZooKeeper operation timeout in seconds. " |
| + "@deprecated - Use metadataStoreOperationTimeoutSeconds instead." |
| ) |
| private int zooKeeperOperationTimeoutSeconds = -1; |
| |
| @Deprecated |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| deprecated = true, |
| doc = "ZooKeeper cache expiry time in seconds. " |
| + "@deprecated - Use metadataStoreCacheExpirySeconds instead." |
| ) |
| private int zooKeeperCacheExpirySeconds = -1; |
| |
| @Deprecated |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| deprecated = true, |
| doc = "Is zooKeeper allow read-only operations." |
| ) |
| private boolean zooKeeperAllowReadOnlyOperations; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Specifies if the function worker should use classloading for validating submissions for built-in " |
| + "connectors and functions. This is required for validateConnectorConfig to take effect. " |
| + "Default is false." |
| ) |
| private Boolean enableClassloadingOfBuiltinFiles = false; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Specifies if the function worker should use classloading for validating submissions for external " |
| + "connectors and functions. This is required for validateConnectorConfig to take effect. " |
| + "Default is false." |
| ) |
| private Boolean enableClassloadingOfExternalFiles = false; |
| |
| @FieldContext( |
| category = CATEGORY_CONNECTORS, |
| doc = "The path to the location to locate builtin connectors" |
| ) |
| private String connectorsDirectory = "./connectors"; |
| @FieldContext( |
| category = CATEGORY_CONNECTORS, |
| doc = "The directory where nar packages are extractors" |
| ) |
| private String narExtractionDirectory = NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR; |
| @FieldContext( |
| category = CATEGORY_CONNECTORS, |
| doc = "Whether to enable referencing connectors directory files by file url in connector (sink/source) " |
| + "creation. Default is true." |
| ) |
| private Boolean enableReferencingConnectorDirectoryFiles = true; |
| @FieldContext( |
| category = CATEGORY_FUNCTIONS, |
| doc = "Regex patterns for enabling creation of connectors by referencing packages in matching http/https " |
| + "urls." |
| ) |
| private List<String> additionalEnabledConnectorUrlPatterns = new ArrayList<>(); |
| @FieldContext( |
| category = CATEGORY_CONNECTORS, |
| doc = "Enables extended validation for connector config with fine-grain annotation based validation " |
| + "during submission. Classloading with either enableClassloadingOfExternalFiles or " |
| + "enableClassloadingOfBuiltinFiles must be enabled on the worker for this to take effect. " |
| + "Default is false." |
| ) |
| private Boolean validateConnectorConfig = false; |
| @FieldContext( |
| category = CATEGORY_FUNCTIONS, |
| doc = "Should the builtin sources/sinks/functions be uploaded for the externally managed runtimes?" |
| ) |
| private Boolean uploadBuiltinSinksSources = true; |
| @FieldContext( |
| category = CATEGORY_FUNCTIONS, |
| doc = "The path to the location to locate builtin functions" |
| ) |
| private String functionsDirectory = "./functions"; |
| @FieldContext( |
| category = CATEGORY_FUNCTIONS, |
| doc = "Whether to enable referencing functions directory files by file url in functions creation. " |
| + "Default is true." |
| ) |
| private Boolean enableReferencingFunctionsDirectoryFiles = true; |
| @FieldContext( |
| category = CATEGORY_FUNCTIONS, |
| doc = "Regex patterns for enabling creation of functions by referencing packages in matching http/https " |
| + "urls." |
| ) |
| private List<String> additionalEnabledFunctionsUrlPatterns = new ArrayList<>(); |
| @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 = "Should the metadata topic be compacted?" |
| ) |
| private Boolean useCompactedMetadataTopic = false; |
| @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 Pulsar 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_PKG, |
| doc = "Flag indicates enabling or disabling function worker using unified PackageManagement service." |
| ) |
| private boolean functionsWorkerEnablePackageManagement = false; |
| @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_STATE, |
| doc = "The implementation class for the state store" |
| ) |
| private String stateStorageProviderImplementation = BKStateStoreProviderImpl.class.getName(); |
| |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The Pulsar topic used for storing function assignment informations" |
| ) |
| private String functionAssignmentTopicName; |
| @FieldContext( |
| category = CATEGORY_FUNC_SCHEDULE_MNG, |
| doc = "The scheduler class used by assigning functions to workers" |
| ) |
| private String schedulerClassName; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The frequency of failure checks, in milliseconds" |
| ) |
| private long failureCheckFreqMs; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The reschedule timeout of function assignment, in milliseconds" |
| ) |
| private long rescheduleTimeoutMs; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The frequency to check whether the cluster needs rebalancing" |
| ) |
| private long rebalanceCheckFreqSec; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "Interval to probe for changes in list of workers, in seconds" |
| ) |
| private int workerListProbeIntervalSec = 60; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_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_RUNTIME_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 = "Whether to enable the broker client authentication used by function workers to talk to brokers" |
| ) |
| private Boolean brokerClientAuthenticationEnabled = null; |
| public boolean isBrokerClientAuthenticationEnabled() { |
| if (brokerClientAuthenticationEnabled != null) { |
| return brokerClientAuthenticationEnabled; |
| } else { |
| return authenticationEnabled; |
| } |
| } |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "The authentication plugin used by function workers to talk to brokers" |
| ) |
| private String brokerClientAuthenticationPlugin; |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "The parameters of the authentication plugin used by function workers to talk to brokers" |
| ) |
| private String brokerClientAuthenticationParameters; |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "Authentication plugin to use when connecting to bookies" |
| ) |
| private String bookkeeperClientAuthenticationPlugin; |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "BookKeeper auth plugin implementation specifics parameters name and values" |
| ) |
| private String bookkeeperClientAuthenticationParametersName; |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "Parameters for bookkeeper auth plugin" |
| ) |
| private String bookkeeperClientAuthenticationParameters; |
| @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; |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Path for the TLS certificate file" |
| ) |
| private String tlsCertificateFilePath; |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Path for the TLS private key file" |
| ) |
| private String tlsKeyFilePath; |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Path for the trusted TLS certificate file" |
| ) |
| private String tlsTrustCertsFilePath = ""; |
| @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", |
| deprecated = true |
| ) |
| // TLS for Functions -> Broker |
| // @deprecated use "pulsar+ssl://" in serviceUrl to enable |
| @Deprecated |
| private boolean useTls = false; |
| @FieldContext( |
| category = CATEGORY_SECURITY, |
| doc = "Whether to enable hostname verification on TLS connections" |
| ) |
| private boolean tlsEnableHostnameVerification = false; |
| @FieldContext( |
| category = CATEGORY_SECURITY, |
| doc = "Tls cert refresh duration in seconds (set 0 to check on every new connection)" |
| ) |
| private long tlsCertRefreshCheckDurationSec = 300; |
| |
| /**** --- KeyStore TLS config variables. --- ****/ |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "Enable TLS with KeyStore type configuration in function worker" |
| ) |
| private boolean tlsEnabledWithKeyStore = false; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "Specify the TLS provider for the function worker service: \n" |
| + "When using TLS authentication with CACert, the valid value is either OPENSSL or JDK.\n" |
| + "When using TLS authentication with KeyStore, available values can be SunJSSE, Conscrypt and etc." |
| ) |
| private String tlsProvider = null; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS KeyStore type configuration in function worker: JKS, PKCS12" |
| ) |
| private String tlsKeyStoreType = "JKS"; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS KeyStore path in function worker" |
| ) |
| private String tlsKeyStore = null; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS KeyStore password for function worker" |
| ) |
| @ToString.Exclude |
| private String tlsKeyStorePassword = null; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS TrustStore type configuration in function worker: JKS, PKCS12" |
| ) |
| private String tlsTrustStoreType = "JKS"; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS TrustStore path in function worker" |
| ) |
| private String tlsTrustStore = null; |
| |
| @FieldContext( |
| category = CATEGORY_KEYSTORE_TLS, |
| doc = "TLS TrustStore password for function worker, null means empty password." |
| ) |
| @ToString.Exclude |
| private String tlsTrustStorePassword = null; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Specify the tls protocols the proxy's web service will use to negotiate during TLS Handshake.\n\n" |
| + "Example:- [TLSv1.3, TLSv1.2]" |
| ) |
| private Set<String> webServiceTlsProtocols = new TreeSet<>(); |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Specify the tls cipher the proxy's web service will use to negotiate during TLS Handshake.\n\n" |
| + "Example:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]" |
| ) |
| private Set<String> webServiceTlsCiphers = new TreeSet<>(); |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Enforce authentication" |
| ) |
| private boolean authenticationEnabled = false; |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Authentication provider name list, which is a list of class names" |
| ) |
| private Set<String> authenticationProviders = new TreeSet<>(); |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Enforce authorization on accessing functions admin-api" |
| ) |
| private boolean authorizationEnabled = false; |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Authorization provider fully qualified class-name" |
| ) |
| private String authorizationProvider = PulsarAuthorizationProvider.class.getName(); |
| @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 = new TreeSet<>(); |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Role names that are treated as `proxy roles`. These are the only roles that can supply the " |
| + "originalPrincipal.") |
| private Set<String> proxyRoles = new TreeSet<>(); |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "This is a regexp, which limits the range of possible ids which can connect to the Broker using SASL." |
| + "\n Default value is: \".*pulsar.*\", so only clients whose id contains 'pulsar' are allowed to" |
| + " connect." |
| ) |
| private String saslJaasClientAllowedIds = SaslConstants.JAAS_CLIENT_ALLOWED_IDS_DEFAULT; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Service Principal, for login context name. Default value is \"PulsarFunction\"." |
| ) |
| private String saslJaasServerSectionName = SaslConstants.JAAS_DEFAULT_FUNCTION_SECTION_NAME; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "Path to file containing the secret to be used to SaslRoleTokenSigner\n" |
| + "The secret can be specified like:\n" |
| + "saslJaasServerRoleTokenSignerSecretPath=file:///my/saslRoleTokenSignerSecret.key." |
| ) |
| private String saslJaasServerRoleTokenSignerSecretPath; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER_SECURITY, |
| doc = "kerberos kinit command." |
| ) |
| private String kinitCommand = "/usr/bin/kinit"; |
| |
| private Properties properties = new Properties(); |
| |
| public boolean getTlsEnabled() { |
| return tlsEnabled && workerPortTls != null; |
| } |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Whether to initialize distributed log metadata in runtime" |
| ) |
| private Boolean initializedDlogMetadata = false; |
| |
| public Boolean isInitializedDlogMetadata() { |
| if (this.initializedDlogMetadata == null){ |
| return false; |
| } |
| return this.initializedDlogMetadata; |
| }; |
| |
| /******** security settings for Pulsar broker client. **********/ |
| |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers" |
| ) |
| private String brokerClientTrustCertsFilePath; |
| |
| public String getBrokerClientTrustCertsFilePath() { |
| // for compatible, if user do not define brokerClientTrustCertsFilePath, we will use tlsTrustCertsFilePath, |
| // otherwise we will use brokerClientTrustCertsFilePath |
| if (StringUtils.isNotBlank(brokerClientTrustCertsFilePath)) { |
| return brokerClientTrustCertsFilePath; |
| } else { |
| return tlsTrustCertsFilePath; |
| } |
| } |
| |
| /******** Function Runtime configurations. **********/ |
| |
| |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The classname of the function runtime factory." |
| ) |
| private String functionRuntimeFactoryClassName; |
| |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "A map of configs for function runtime factory." |
| ) |
| private Map<String, Object> functionRuntimeFactoryConfigs; |
| |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The classname of the secrets provider configurator." |
| ) |
| private String secretsProviderConfiguratorClassName; |
| @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 SecretsProviderConfigurator" |
| ) |
| private Map<String, String> secretsProviderConfiguratorConfig; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "A set of the minimum amount of resources functions must request. " |
| + "Support for this depends on function runtime." |
| ) |
| private Resources functionInstanceMinResources; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "A set of the maximum amount of resources functions may request. " |
| + "Support for this depends on function runtime." |
| ) |
| private Resources functionInstanceMaxResources; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "Granularities of requested resources. If the granularity of any type of resource is set," |
| + " the requested resource of the type must be a multiple of the granularity." |
| ) |
| private Resources functionInstanceResourceGranularities; |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "If this configuration is set to be true, the amount of requested resources of all type of resources" |
| + " that have the granularity set must be the same multiples of their granularities." |
| ) |
| private boolean functionInstanceResourceChangeInLockStep = false; |
| |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "The class name of the Function Authentication Provider to use." |
| + " The Function Authentication Provider is responsible to distributing the necessary" |
| + " authentication information to individual functions e.g. user tokens" |
| ) |
| @Getter(AccessLevel.NONE) |
| private String functionAuthProviderClassName; |
| |
| public String getFunctionAuthProviderClassName() { |
| // if we haven't set a value and are running kubernetes, we default to the SecretsTokenAuthProvider |
| // as that matches behavior before this property could be overridden |
| if (!StringUtils.isEmpty(functionAuthProviderClassName)) { |
| return functionAuthProviderClassName; |
| } else { |
| if (StringUtils.equals(this.getFunctionRuntimeFactoryClassName(), KubernetesRuntimeFactory.class.getName()) |
| || getKubernetesContainerFactory() != null) { |
| return KubernetesSecretsTokenAuthProvider.class.getName(); |
| } |
| return null; |
| } |
| } |
| |
| @FieldContext( |
| doc = "The full class-name of an instance of RuntimeCustomizer." |
| + " This class receives the 'customRuntimeOptions string and can customize" |
| + " details of how the runtime operates" |
| ) |
| protected String runtimeCustomizerClassName; |
| |
| @FieldContext( |
| doc = "A map of config passed to the RuntimeCustomizer." |
| + " This config is distinct from the `customRuntimeOptions` provided by functions" |
| + " as this config is the the same across all functions" |
| ) |
| private Map<String, Object> runtimeCustomizerConfig = Collections.emptyMap(); |
| |
| @FieldContext( |
| doc = "Max pending async requests per instance to avoid large number of concurrent requests." |
| + "Only used in AsyncFunction. Default: 1000" |
| ) |
| private int maxPendingAsyncRequests = 1000; |
| |
| @FieldContext( |
| doc = "Whether to forward the source message properties to the output message" |
| ) |
| private boolean forwardSourceMessageProperty = true; |
| |
| @FieldContext( |
| doc = "Additional arguments to pass to the Java command line for Java functions" |
| ) |
| private List<String> additionalJavaRuntimeArguments = new ArrayList<>(); |
| |
| @FieldContext( |
| category = CATEGORY_CONNECTORS, |
| doc = "Whether to ignore unknown properties when deserializing the connector configuration. " |
| + "After upgrading a connector to a new version with a new configuration, " |
| + "the new configuration may not be compatible with the old connector. " |
| + "In case of rollback, it's required to also rollback the connector configuration. " |
| + "Ignoring unknown fields makes possible to keep the new configuration and " |
| + "only rollback the connector." |
| ) |
| private boolean ignoreUnknownConfigFields = false; |
| |
| public String getFunctionMetadataTopic() { |
| return String.format("persistent://%s/%s", pulsarFunctionsNamespace, functionMetadataTopicName); |
| } |
| |
| public String getClusterCoordinationTopic() { |
| return String.format("persistent://%s/%s", pulsarFunctionsNamespace, clusterCoordinationTopicName); |
| } |
| |
| public String getFunctionAssignmentTopic() { |
| return String.format("persistent://%s/%s", pulsarFunctionsNamespace, functionAssignmentTopicName); |
| } |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "The nar package for the function worker service" |
| ) |
| private String functionsWorkerServiceNarPackage = ""; |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "The additional configs for the function worker service if functionsWorkerServiceNarPackage provided" |
| ) |
| private Map<String, Object> functionsWorkerServiceCustomConfigs = Collections.emptyMap(); |
| |
| @FieldContext( |
| category = CATEGORY_WORKER, |
| doc = "Enable to expose Pulsar Admin Client from Function Context, default is disabled" |
| ) |
| private boolean exposeAdminClientEnabled = false; |
| |
| public static WorkerConfig load(String yamlFile) throws IOException { |
| if (isBlank(yamlFile)) { |
| return new WorkerConfig(); |
| } |
| ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); |
| return mapper.readValue(new File(yamlFile), WorkerConfig.class); |
| } |
| |
| public String getWorkerId() { |
| if (isBlank(this.workerId)) { |
| this.workerId = String.format("%s-%s", this.getWorkerHostname(), this.getWorkerPort() != null |
| ? this.getWorkerPort() : this.getWorkerPortTls()); |
| } |
| return this.workerId; |
| } |
| |
| public String getWorkerHostname() { |
| if (isBlank(this.workerHostname)) { |
| this.workerHostname = unsafeLocalhostResolve(); |
| } |
| return this.workerHostname; |
| } |
| |
| public byte[] getTlsTrustChainBytes() { |
| if (StringUtils.isNotEmpty(getBrokerClientTrustCertsFilePath()) |
| && Files.exists(Paths.get(getBrokerClientTrustCertsFilePath()))) { |
| try { |
| return Files.readAllBytes(Paths.get(getBrokerClientTrustCertsFilePath())); |
| } catch (IOException e) { |
| throw new IllegalStateException("Failed to read CA bytes", e); |
| } |
| } else { |
| return null; |
| } |
| } |
| |
| public String getWorkerWebAddress() { |
| return String.format("http://%s:%d", this.getWorkerHostname(), this.getWorkerPort()); |
| } |
| |
| public String getWorkerWebAddressTls() { |
| return String.format("https://%s:%d", this.getWorkerHostname(), this.getWorkerPortTls()); |
| } |
| |
| public static String unsafeLocalhostResolve() { |
| try { |
| // Get the fully qualified hostname |
| return InetAddress.getLocalHost().getCanonicalHostName(); |
| } catch (UnknownHostException ex) { |
| throw new IllegalStateException("Failed to resolve localhost name.", ex); |
| } |
| } |
| |
| public String getConfigurationMetadataStoreUrl() { |
| if (StringUtils.isNotBlank(configurationMetadataStoreUrl)) { |
| return configurationMetadataStoreUrl; |
| } else { |
| return configurationStoreServers; |
| } |
| } |
| |
| @Override |
| public void setProperties(Properties properties) { |
| this.properties = properties; |
| } |
| |
| /********* DEPRECATED CONFIGS. *********/ |
| |
| @Deprecated |
| @Data |
| /** |
| * @Deprecated in favor for using functionRuntimeFactoryClassName and functionRuntimeFactoryConfigs |
| * for specifying the function runtime and configs to use |
| */ |
| public static class ThreadContainerFactory extends ThreadRuntimeFactoryConfig { |
| |
| } |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "Thread based runtime settings" |
| ) |
| @Deprecated |
| private ThreadContainerFactory threadContainerFactory; |
| |
| @Deprecated |
| @Data |
| /** |
| * @Deprecated in favor for using functionRuntimeFactoryClassName and functionRuntimeFactoryConfigs |
| * for specifying the function runtime and configs to use |
| */ |
| public static class ProcessContainerFactory extends ProcessRuntimeFactoryConfig { |
| |
| } |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "Process based runtime settings" |
| ) |
| @Deprecated |
| private ProcessContainerFactory processContainerFactory; |
| |
| @Deprecated |
| @Data |
| /** |
| * @Deprecated in favor for using functionRuntimeFactoryClassName and functionRuntimeFactoryConfigs |
| * for specifying the function runtime and configs to use |
| */ |
| public static class KubernetesContainerFactory extends KubernetesRuntimeFactoryConfig { |
| |
| } |
| @FieldContext( |
| category = CATEGORY_FUNC_RUNTIME_MNG, |
| doc = "Kubernetes based runtime settings" |
| ) |
| @Deprecated |
| private KubernetesContainerFactory kubernetesContainerFactory; |
| |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "The parameters of the authentication plugin used by function workers to talk to brokers" |
| ) |
| @Deprecated |
| private String clientAuthenticationParameters; |
| @FieldContext( |
| category = CATEGORY_CLIENT_SECURITY, |
| doc = "The authentication plugin used by function workers to talk to brokers" |
| ) |
| @Deprecated |
| private String clientAuthenticationPlugin; |
| |
| public String getBrokerClientAuthenticationPlugin() { |
| if (null == brokerClientAuthenticationPlugin) { |
| return clientAuthenticationPlugin; |
| } else { |
| return brokerClientAuthenticationPlugin; |
| } |
| } |
| |
| public String getBrokerClientAuthenticationParameters() { |
| if (null == brokerClientAuthenticationParameters) { |
| return clientAuthenticationParameters; |
| } else { |
| return brokerClientAuthenticationParameters; |
| } |
| } |
| |
| public long getMetadataStoreSessionTimeoutMillis() { |
| return zooKeeperSessionTimeoutMillis > 0 ? zooKeeperSessionTimeoutMillis : metadataStoreSessionTimeoutMillis; |
| } |
| |
| public int getMetadataStoreOperationTimeoutSeconds() { |
| return zooKeeperOperationTimeoutSeconds > 0 ? zooKeeperOperationTimeoutSeconds |
| : metadataStoreOperationTimeoutSeconds; |
| } |
| |
| public int getMetadataStoreCacheExpirySeconds() { |
| return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds; |
| } |
| |
| public boolean isMetadataStoreAllowReadOnlyOperations() { |
| return zooKeeperAllowReadOnlyOperations || metadataStoreAllowReadOnlyOperations; |
| } |
| } |
