| /* |
| * 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.nifi.minifi.bootstrap.util; |
| |
| |
| import org.apache.commons.io.output.ByteArrayOutputStream; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException; |
| import org.apache.nifi.minifi.bootstrap.exception.InvalidConfigurationException; |
| import org.apache.nifi.minifi.commons.schema.ComponentStatusRepositorySchema; |
| import org.apache.nifi.minifi.commons.schema.ConfigSchema; |
| import org.apache.nifi.minifi.commons.schema.ConnectionSchema; |
| import org.apache.nifi.minifi.commons.schema.ContentRepositorySchema; |
| import org.apache.nifi.minifi.commons.schema.ControllerServiceSchema; |
| import org.apache.nifi.minifi.commons.schema.CorePropertiesSchema; |
| import org.apache.nifi.minifi.commons.schema.FlowControllerSchema; |
| import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema; |
| import org.apache.nifi.minifi.commons.schema.FunnelSchema; |
| import org.apache.nifi.minifi.commons.schema.PortSchema; |
| import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema; |
| import org.apache.nifi.minifi.commons.schema.ProcessorSchema; |
| import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema; |
| import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema; |
| import org.apache.nifi.minifi.commons.schema.RemotePortSchema; |
| import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema; |
| import org.apache.nifi.minifi.commons.schema.ReportingSchema; |
| import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema; |
| import org.apache.nifi.minifi.commons.schema.SensitivePropsSchema; |
| import org.apache.nifi.minifi.commons.schema.SwapSchema; |
| import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema; |
| import org.apache.nifi.minifi.commons.schema.common.Schema; |
| import org.apache.nifi.minifi.commons.schema.common.StringUtil; |
| import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.w3c.dom.DOMException; |
| import org.w3c.dom.Document; |
| import org.w3c.dom.Element; |
| |
| import javax.xml.parsers.DocumentBuilder; |
| import javax.xml.parsers.DocumentBuilderFactory; |
| import javax.xml.parsers.ParserConfigurationException; |
| import javax.xml.transform.OutputKeys; |
| import javax.xml.transform.Transformer; |
| import javax.xml.transform.TransformerException; |
| import javax.xml.transform.TransformerFactory; |
| import javax.xml.transform.TransformerFactoryConfigurationError; |
| import javax.xml.transform.dom.DOMSource; |
| import javax.xml.transform.stream.StreamResult; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.nio.file.Files; |
| import java.nio.file.Path; |
| import java.nio.file.Paths; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.stream.Collectors; |
| import java.util.zip.GZIPOutputStream; |
| |
| public final class ConfigTransformer { |
| // Underlying version of NIFI will be using |
| public static final String NIFI_VERSION = "1.8.0"; |
| public static final String ROOT_GROUP = "Root-Group"; |
| public static final String NIFI_VERSION_KEY = "nifi.version"; |
| |
| public static final Logger logger = LoggerFactory.getLogger(ConfigTransformer.class); |
| |
| // Final util classes should have private constructor |
| private ConfigTransformer() { |
| } |
| |
| public static void transformConfigFile(InputStream sourceStream, String destPath, Properties bootstrapProperties) throws Exception { |
| ConvertableSchema<ConfigSchema> convertableSchema = throwIfInvalid(SchemaLoader.loadConvertableSchemaFromYaml(sourceStream)); |
| ConfigSchema configSchema = throwIfInvalid(convertableSchema.convert()); |
| |
| SecurityPropertiesSchema securityProperties = BootstrapTransformer.buildSecurityPropertiesFromBootstrap(bootstrapProperties).orElse(null); |
| ProvenanceReportingSchema provenanceReportingProperties = BootstrapTransformer.buildProvenanceReportingPropertiesFromBootstrap(bootstrapProperties).orElse(null); |
| |
| // See if we are providing defined properties from the filesystem configurations and use those as the definitive values |
| if (securityProperties != null) { |
| configSchema.setSecurityProperties(securityProperties); |
| logger.info("Bootstrap flow override: Replaced security properties"); |
| } |
| if (provenanceReportingProperties != null) { |
| configSchema.setProvenanceReportingProperties(provenanceReportingProperties); |
| logger.info("Bootstrap flow override: Replaced provenance reporting properties"); |
| } |
| |
| // Replace all processor SSL controller services with MiNiFi parent, if bootstrap boolean is set to true |
| if (BootstrapTransformer.processorSSLOverride(bootstrapProperties)) { |
| for (ProcessorSchema processorConfig : configSchema.getProcessGroupSchema().getProcessors()) { |
| processorConfig.getProperties().replace("SSL Context Service", processorConfig.getProperties().get("SSL Context Service"), "SSL-Context-Service"); |
| logger.info("Bootstrap flow override: Replaced {} SSL Context Service with parent MiNiFi SSL", processorConfig.getName()); |
| } |
| } |
| |
| // Create nifi.properties and flow.xml.gz in memory |
| ByteArrayOutputStream nifiPropertiesOutputStream = new ByteArrayOutputStream(); |
| writeNiFiProperties(configSchema, nifiPropertiesOutputStream); |
| |
| writeFlowXmlFile(configSchema, destPath); |
| |
| // Write nifi.properties and flow.xml.gz |
| writeNiFiPropertiesFile(nifiPropertiesOutputStream, destPath); |
| } |
| |
| private static <T extends Schema> T throwIfInvalid(T schema) throws InvalidConfigurationException { |
| if (!schema.isValid()) { |
| throw new InvalidConfigurationException("Failed to transform config file due to:[" |
| + schema.getValidationIssues().stream().sorted().collect(Collectors.joining("], [")) + "]"); |
| } |
| return schema; |
| } |
| |
| protected static void writeNiFiPropertiesFile(ByteArrayOutputStream nifiPropertiesOutputStream, String destPath) throws IOException { |
| final Path nifiPropertiesPath = Paths.get(destPath, "nifi.properties"); |
| try (FileOutputStream nifiProperties = new FileOutputStream(nifiPropertiesPath.toString())) { |
| nifiPropertiesOutputStream.writeTo(nifiProperties); |
| } finally { |
| if (nifiPropertiesOutputStream != null) { |
| nifiPropertiesOutputStream.flush(); |
| nifiPropertiesOutputStream.close(); |
| } |
| } |
| } |
| |
| protected static void writeFlowXmlFile(ConfigSchema configSchema, OutputStream outputStream) throws TransformerException, ConfigTransformerException, ConfigurationChangeException, IOException { |
| final StreamResult streamResult = new StreamResult(outputStream); |
| |
| // configure the transformer and convert the DOM |
| final TransformerFactory transformFactory = TransformerFactory.newInstance(); |
| final Transformer transformer = transformFactory.newTransformer(); |
| transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount", "2"); |
| transformer.setOutputProperty(OutputKeys.INDENT, "yes"); |
| |
| // transform the document to byte stream |
| transformer.transform(createFlowXml(configSchema), streamResult); |
| } |
| |
| protected static void writeFlowXmlFile(ConfigSchema configSchema, String path) throws IOException, TransformerException, ConfigurationChangeException, ConfigTransformerException { |
| try (OutputStream fileOut = Files.newOutputStream(Paths.get(path, "flow.xml.gz"))) { |
| try (OutputStream outStream = new GZIPOutputStream(fileOut)) { |
| writeFlowXmlFile(configSchema, outStream); |
| } |
| } |
| } |
| |
| protected static void writeNiFiProperties(ConfigSchema configSchema, OutputStream outputStream) throws IOException, ConfigurationChangeException { |
| try { |
| CorePropertiesSchema coreProperties = configSchema.getCoreProperties(); |
| FlowFileRepositorySchema flowfileRepoSchema = configSchema.getFlowfileRepositoryProperties(); |
| SwapSchema swapProperties = flowfileRepoSchema.getSwapProperties(); |
| ContentRepositorySchema contentRepoProperties = configSchema.getContentRepositoryProperties(); |
| ComponentStatusRepositorySchema componentStatusRepoProperties = configSchema.getComponentStatusRepositoryProperties(); |
| SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties(); |
| SensitivePropsSchema sensitiveProperties = securityProperties.getSensitiveProps(); |
| ProvenanceRepositorySchema provenanceRepositorySchema = configSchema.getProvenanceRepositorySchema(); |
| |
| OrderedProperties orderedProperties = new OrderedProperties(); |
| orderedProperties.setProperty(NIFI_VERSION_KEY, NIFI_VERSION, "# Core Properties #" + System.lineSeparator()); |
| orderedProperties.setProperty("nifi.flow.configuration.file", "./conf/flow.xml.gz"); |
| orderedProperties.setProperty("nifi.flow.configuration.archive.enabled", "false"); |
| orderedProperties.setProperty("nifi.flow.configuration.archive.dir", "./conf/archive/"); |
| orderedProperties.setProperty("nifi.flowcontroller.autoResumeState", "true"); |
| orderedProperties.setProperty("nifi.flowcontroller.graceful.shutdown.period", coreProperties.getFlowControllerGracefulShutdownPeriod()); |
| orderedProperties.setProperty("nifi.flowservice.writedelay.interval", coreProperties.getFlowServiceWriteDelayInterval()); |
| orderedProperties.setProperty("nifi.administrative.yield.duration", coreProperties.getAdministrativeYieldDuration()); |
| orderedProperties.setProperty("nifi.variable.registry.properties", coreProperties.getVariableRegistryProperties()); |
| |
| orderedProperties.setProperty("nifi.bored.yield.duration", coreProperties.getBoredYieldDuration(), |
| "# If a component has no work to do (is \"bored\"), how long should we wait before checking again for work?"); |
| |
| orderedProperties.setProperty("nifi.authority.provider.configuration.file", "./conf/authority-providers.xml", ""); |
| orderedProperties.setProperty("nifi.login.identity.provider.configuration.file", "./conf/login-identity-providers.xml"); |
| orderedProperties.setProperty("nifi.templates.directory", "./conf/templates"); |
| orderedProperties.setProperty("nifi.ui.banner.text", ""); |
| orderedProperties.setProperty("nifi.ui.autorefresh.interval", "30 sec"); |
| orderedProperties.setProperty("nifi.nar.library.directory", "./lib"); |
| orderedProperties.setProperty("nifi.nar.working.directory", "./work/nar/"); |
| orderedProperties.setProperty("nifi.documentation.working.directory", "./work/docs/components"); |
| |
| orderedProperties.setProperty("nifi.state.management.configuration.file", "./conf/state-management.xml", System.lineSeparator() + |
| "####################" + |
| "# State Management #" + |
| "####################"); |
| |
| orderedProperties.setProperty("nifi.state.management.provider.local", "local-provider", "# The ID of the local state provider"); |
| |
| orderedProperties.setProperty("nifi.database.directory", "./database_repository", System.lineSeparator() + "# H2 Settings"); |
| orderedProperties.setProperty("nifi.h2.url.append", ";LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE"); |
| orderedProperties.setProperty("nifi.flowfile.repository.implementation", "org.apache.nifi.controller.repository.WriteAheadFlowFileRepository", |
| System.lineSeparator() + "# FlowFile Repository"); |
| orderedProperties.setProperty("nifi.flowfile.repository.directory", "./flowfile_repository"); |
| orderedProperties.setProperty("nifi.flowfile.repository.partitions", String.valueOf(flowfileRepoSchema.getPartitions())); |
| orderedProperties.setProperty("nifi.flowfile.repository.checkpoint.interval", flowfileRepoSchema.getCheckpointInterval()); |
| orderedProperties.setProperty("nifi.flowfile.repository.always.sync", Boolean.toString(flowfileRepoSchema.getAlwaysSync())); |
| |
| orderedProperties.setProperty("nifi.swap.manager.implementation", "org.apache.nifi.controller.FileSystemSwapManager", ""); |
| orderedProperties.setProperty("nifi.queue.swap.threshold", String.valueOf(swapProperties.getThreshold())); |
| orderedProperties.setProperty("nifi.swap.in.period", swapProperties.getInPeriod()); |
| orderedProperties.setProperty("nifi.swap.in.threads", String.valueOf(swapProperties.getInThreads())); |
| orderedProperties.setProperty("nifi.swap.out.period", swapProperties.getOutPeriod()); |
| orderedProperties.setProperty("nifi.swap.out.threads", String.valueOf(swapProperties.getOutThreads())); |
| |
| orderedProperties.setProperty("nifi.content.repository.implementation", contentRepoProperties.getContentRepository(), System.lineSeparator() + "# Content Repository"); |
| orderedProperties.setProperty("nifi.content.claim.max.appendable.size", contentRepoProperties.getContentClaimMaxAppendableSize()); |
| orderedProperties.setProperty("nifi.content.claim.max.flow.files", String.valueOf(contentRepoProperties.getContentClaimMaxFlowFiles())); |
| orderedProperties.setProperty("nifi.content.repository.archive.max.retention.period", contentRepoProperties.getContentRepoArchiveMaxRetentionPeriod()); |
| orderedProperties.setProperty("nifi.content.repository.archive.max.usage.percentage", contentRepoProperties.getContentRepoArchiveMaxUsagePercentage()); |
| orderedProperties.setProperty("nifi.content.repository.archive.enabled", Boolean.toString(contentRepoProperties.getContentRepoArchiveEnabled())); |
| orderedProperties.setProperty("nifi.content.repository.directory.default", "./content_repository"); |
| orderedProperties.setProperty("nifi.content.repository.always.sync", Boolean.toString(contentRepoProperties.getAlwaysSync())); |
| |
| orderedProperties.setProperty("nifi.provenance.repository.implementation", provenanceRepositorySchema.getProvenanceRepository(), |
| System.lineSeparator() + "# Provenance Repository Properties"); |
| |
| orderedProperties.setProperty("nifi.provenance.repository.rollover.time", provenanceRepositorySchema.getProvenanceRepoRolloverTimeKey()); |
| |
| orderedProperties.setProperty("nifi.provenance.repository.index.shard.size", provenanceRepositorySchema.getProvenanceRepoIndexShardSize()); |
| orderedProperties.setProperty("nifi.provenance.repository.max.storage.size", provenanceRepositorySchema.getProvenanceRepoMaxStorageSize()); |
| orderedProperties.setProperty("nifi.provenance.repository.max.storage.time", provenanceRepositorySchema.getProvenanceRepoMaxStorageTime()); |
| |
| orderedProperties.setProperty("nifi.provenance.repository.buffer.size", String.valueOf(provenanceRepositorySchema.getProvenanceRepoBufferSize()), |
| System.lineSeparator() + "# Volatile Provenance Respository Properties"); |
| |
| orderedProperties.setProperty("nifi.components.status.repository.implementation", "org.apache.nifi.controller.status.history.VolatileComponentStatusRepository", |
| System.lineSeparator() + "# Component Status Repository"); |
| orderedProperties.setProperty("nifi.components.status.repository.buffer.size", String.valueOf(componentStatusRepoProperties.getBufferSize())); |
| orderedProperties.setProperty("nifi.components.status.snapshot.frequency", componentStatusRepoProperties.getSnapshotFrequency()); |
| |
| orderedProperties.setProperty("nifi.web.war.directory", "./lib", System.lineSeparator() + "# web properties #"); |
| orderedProperties.setProperty("nifi.web.http.host", ""); |
| orderedProperties.setProperty("nifi.web.http.port", "8081"); |
| orderedProperties.setProperty("nifi.web.https.host", ""); |
| orderedProperties.setProperty("nifi.web.https.port", ""); |
| orderedProperties.setProperty("nifi.web.jetty.working.directory", "./work/jetty"); |
| orderedProperties.setProperty("nifi.web.jetty.threads", "200"); |
| |
| final String sensitivePropertiesKey = sensitiveProperties.getKey(); |
| final String notnullSensitivePropertiesKey = sensitivePropertiesKey != null ? sensitivePropertiesKey : ""; |
| orderedProperties.setProperty("nifi.sensitive.props.key", notnullSensitivePropertiesKey, System.lineSeparator() + "# security properties #"); |
| orderedProperties.setProperty("nifi.sensitive.props.algorithm", sensitiveProperties.getAlgorithm()); |
| orderedProperties.setProperty("nifi.sensitive.props.provider", sensitiveProperties.getProvider()); |
| |
| orderedProperties.setProperty("nifi.security.keystore", securityProperties.getKeystore(), ""); |
| orderedProperties.setProperty("nifi.security.keystoreType", securityProperties.getKeystoreType()); |
| orderedProperties.setProperty("nifi.security.keystorePasswd", securityProperties.getKeystorePassword()); |
| orderedProperties.setProperty("nifi.security.keyPasswd", securityProperties.getKeyPassword()); |
| orderedProperties.setProperty("nifi.security.truststore", securityProperties.getTruststore()); |
| orderedProperties.setProperty("nifi.security.truststoreType", securityProperties.getTruststoreType()); |
| orderedProperties.setProperty("nifi.security.truststorePasswd", securityProperties.getTruststorePassword()); |
| orderedProperties.setProperty("nifi.security.needClientAuth", ""); |
| orderedProperties.setProperty("nifi.security.user.credential.cache.duration", "24 hours"); |
| orderedProperties.setProperty("nifi.security.user.authority.provider", "file-provider"); |
| orderedProperties.setProperty("nifi.security.user.login.identity.provider", ""); |
| orderedProperties.setProperty("nifi.security.support.new.account.requests", ""); |
| |
| orderedProperties.setProperty("nifi.security.anonymous.authorities", "", "# Valid Authorities include: ROLE_MONITOR,ROLE_DFM,ROLE_ADMIN,ROLE_PROVENANCE,ROLE_NIFI"); |
| orderedProperties.setProperty("nifi.security.ocsp.responder.url", ""); |
| orderedProperties.setProperty("nifi.security.ocsp.responder.certificate", ""); |
| |
| orderedProperties.setProperty("nifi.cluster.is.node", "false", System.lineSeparator() + System.lineSeparator() + "# cluster node properties (only configure for cluster nodes) #"); |
| orderedProperties.setProperty("nifi.cluster.is.manager", "false", System.lineSeparator() + "# cluster manager properties (only configure for cluster manager) #"); |
| |
| for (Map.Entry<String, String> entry : configSchema.getNifiPropertiesOverrides().entrySet()) { |
| orderedProperties.setProperty(entry.getKey(), entry.getValue()); |
| } |
| |
| orderedProperties.store(outputStream, PROPERTIES_FILE_APACHE_2_0_LICENSE); |
| } catch (NullPointerException e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while creating the nifi.properties", e); |
| } finally { |
| outputStream.close(); |
| } |
| } |
| |
| protected static DOMSource createFlowXml(ConfigSchema configSchema) throws IOException, ConfigurationChangeException, ConfigTransformerException { |
| try { |
| // create a new, empty document |
| final DocumentBuilderFactory docFactory = DocumentBuilderFactory.newInstance(); |
| docFactory.setNamespaceAware(true); |
| |
| final DocumentBuilder docBuilder = docFactory.newDocumentBuilder(); |
| final Document doc = docBuilder.newDocument(); |
| |
| // populate document with controller state |
| final Element rootNode = doc.createElement("flowController"); |
| doc.appendChild(rootNode); |
| CorePropertiesSchema coreProperties = configSchema.getCoreProperties(); |
| addTextElement(rootNode, "maxTimerDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads())); |
| addTextElement(rootNode, "maxEventDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads())); |
| |
| FlowControllerSchema flowControllerProperties = configSchema.getFlowControllerProperties(); |
| |
| final Element element = doc.createElement("rootGroup"); |
| rootNode.appendChild(element); |
| |
| ProcessGroupSchema processGroupSchema = configSchema.getProcessGroupSchema(); |
| processGroupSchema.setId(ROOT_GROUP); |
| processGroupSchema.setName(flowControllerProperties.getName()); |
| processGroupSchema.setComment(flowControllerProperties.getComment()); |
| |
| addProcessGroup(doc, element, processGroupSchema, new ParentGroupIdResolver(processGroupSchema)); |
| |
| SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties(); |
| boolean useSSL = securityProperties.useSSL(); |
| if (useSSL) { |
| Element controllerServicesNode = doc.getElementById("controllerServices"); |
| if (controllerServicesNode == null) { |
| controllerServicesNode = doc.createElement("controllerServices"); |
| } |
| |
| rootNode.appendChild(controllerServicesNode); |
| addSSLControllerService(controllerServicesNode, securityProperties); |
| } |
| |
| List<ReportingSchema> reportingTasks = configSchema.getReportingTasksSchema(); |
| ProvenanceReportingSchema provenanceProperties = configSchema.getProvenanceReportingProperties(); |
| if (provenanceProperties != null) { |
| provenanceProperties.setSSL(useSSL); |
| ReportingSchema provenance = provenanceProperties.convert(); |
| provenance.setId("Provenance-Reporting"); |
| provenance.setName("Site-To-Site-Provenance-Reporting"); |
| reportingTasks.add(provenance); |
| } |
| if (reportingTasks != null) { |
| final Element reportingTasksNode = doc.createElement("reportingTasks"); |
| rootNode.appendChild(reportingTasksNode); |
| for (ReportingSchema task : reportingTasks) { |
| addReportingTask(reportingTasksNode, task); |
| } |
| } |
| |
| return new DOMSource(doc); |
| } catch (final ParserConfigurationException | DOMException | TransformerFactoryConfigurationError | IllegalArgumentException e) { |
| throw new ConfigTransformerException(e); |
| } catch (Exception e) { |
| throw new ConfigTransformerException("Failed to parse the config YAML while writing the top level of the flow xml", e); |
| } |
| } |
| |
| protected static void addSSLControllerService(final Element element, SecurityPropertiesSchema securityProperties) throws ConfigurationChangeException { |
| try { |
| final Element serviceElement = element.getOwnerDocument().createElement("controllerService"); |
| addTextElement(serviceElement, "id", "SSL-Context-Service"); |
| addTextElement(serviceElement, "name", "SSL-Context-Service"); |
| addTextElement(serviceElement, "comment", ""); |
| addTextElement(serviceElement, "class", "org.apache.nifi.ssl.StandardSSLContextService"); |
| |
| addTextElement(serviceElement, "enabled", "true"); |
| |
| Map<String, Object> attributes = new HashMap<>(); |
| attributes.put("Keystore Filename", securityProperties.getKeystore()); |
| attributes.put("Keystore Type", securityProperties.getKeystoreType()); |
| attributes.put("Keystore Password", securityProperties.getKeystorePassword()); |
| attributes.put("Truststore Filename", securityProperties.getTruststore()); |
| attributes.put("Truststore Type", securityProperties.getTruststoreType()); |
| attributes.put("Truststore Password", securityProperties.getTruststorePassword()); |
| attributes.put("SSL Protocol", securityProperties.getSslProtocol()); |
| |
| addConfiguration(serviceElement, attributes); |
| |
| element.appendChild(serviceElement); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to create an SSL Controller Service", e); |
| } |
| } |
| |
| protected static void addControllerService(final Element element, ControllerServiceSchema controllerServiceSchema) throws ConfigurationChangeException { |
| try { |
| final Element serviceElement = element.getOwnerDocument().createElement("controllerService"); |
| addTextElement(serviceElement, "id", controllerServiceSchema.getId()); |
| addTextElement(serviceElement, "name", controllerServiceSchema.getName()); |
| addTextElement(serviceElement, "comment", ""); |
| addTextElement(serviceElement, "class", controllerServiceSchema.getServiceClass()); |
| |
| addTextElement(serviceElement, "enabled", "true"); |
| |
| Map<String, Object> attributes = controllerServiceSchema.getProperties(); |
| |
| addConfiguration(serviceElement, attributes); |
| |
| String annotationData = controllerServiceSchema.getAnnotationData(); |
| if (annotationData != null && !annotationData.isEmpty()) { |
| addTextElement(element, "annotationData", annotationData); |
| } |
| |
| element.appendChild(serviceElement); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to create an SSL Controller Service", e); |
| } |
| } |
| |
| protected static void addProcessGroup(Document doc, Element element, ProcessGroupSchema processGroupSchema, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException { |
| try { |
| String processGroupId = processGroupSchema.getId(); |
| addTextElement(element, "id", processGroupId); |
| addTextElement(element, "name", processGroupSchema.getName()); |
| addPosition(element); |
| addTextElement(element, "comment", processGroupSchema.getComment()); |
| |
| for (ProcessorSchema processorConfig : processGroupSchema.getProcessors()) { |
| addProcessor(element, processorConfig); |
| } |
| |
| for (PortSchema portSchema : processGroupSchema.getInputPortSchemas()) { |
| addPort(doc, element, portSchema, "inputPort"); |
| } |
| |
| for (PortSchema portSchema : processGroupSchema.getOutputPortSchemas()) { |
| addPort(doc, element, portSchema, "outputPort"); |
| } |
| |
| for (FunnelSchema funnelSchema : processGroupSchema.getFunnels()) { |
| addFunnel(element, funnelSchema); |
| } |
| |
| for (ProcessGroupSchema child : processGroupSchema.getProcessGroupSchemas()) { |
| Element processGroups = doc.createElement("processGroup"); |
| element.appendChild(processGroups); |
| addProcessGroup(doc, processGroups, child, parentGroupIdResolver); |
| } |
| |
| for (RemoteProcessGroupSchema remoteProcessGroupSchema : processGroupSchema.getRemoteProcessGroups()) { |
| addRemoteProcessGroup(element, remoteProcessGroupSchema); |
| } |
| |
| for (ConnectionSchema connectionConfig : processGroupSchema.getConnections()) { |
| addConnection(element, connectionConfig, parentGroupIdResolver); |
| } |
| |
| for (ControllerServiceSchema controllerServiceSchema : processGroupSchema.getControllerServices()) { |
| addControllerService(element, controllerServiceSchema); |
| } |
| } catch (ConfigurationChangeException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to creating the root Process Group", e); |
| } |
| } |
| |
| protected static void addPort(Document doc, Element parentElement, PortSchema portSchema, String tag) { |
| Element element = doc.createElement(tag); |
| parentElement.appendChild(element); |
| |
| addTextElement(element, "id", portSchema.getId()); |
| addTextElement(element, "name", portSchema.getName()); |
| |
| addPosition(element); |
| addTextElement(element, "comments", null); |
| |
| addTextElement(element, "scheduledState", "RUNNING"); |
| } |
| |
| protected static void addProcessor(final Element parentElement, ProcessorSchema processorConfig) throws ConfigurationChangeException { |
| try { |
| final Document doc = parentElement.getOwnerDocument(); |
| final Element element = doc.createElement("processor"); |
| parentElement.appendChild(element); |
| |
| addTextElement(element, "id", processorConfig.getId()); |
| addTextElement(element, "name", processorConfig.getName()); |
| |
| addPosition(element); |
| addStyle(element); |
| |
| addTextElement(element, "comment", ""); |
| addTextElement(element, "class", processorConfig.getProcessorClass()); |
| addTextElement(element, "maxConcurrentTasks", String.valueOf(processorConfig.getMaxConcurrentTasks())); |
| addTextElement(element, "schedulingPeriod", processorConfig.getSchedulingPeriod()); |
| addTextElement(element, "penalizationPeriod", processorConfig.getPenalizationPeriod()); |
| addTextElement(element, "yieldPeriod", processorConfig.getYieldPeriod()); |
| addTextElement(element, "bulletinLevel", "WARN"); |
| addTextElement(element, "lossTolerant", "false"); |
| addTextElement(element, "scheduledState", "RUNNING"); |
| addTextElement(element, "schedulingStrategy", processorConfig.getSchedulingStrategy()); |
| addTextElement(element, "runDurationNanos", String.valueOf(processorConfig.getRunDurationNanos())); |
| |
| String annotationData = processorConfig.getAnnotationData(); |
| if (annotationData != null && !annotationData.isEmpty()) { |
| addTextElement(element, "annotationData", annotationData); |
| } |
| |
| addConfiguration(element, processorConfig.getProperties()); |
| |
| Collection<String> autoTerminatedRelationships = processorConfig.getAutoTerminatedRelationshipsList(); |
| if (autoTerminatedRelationships != null) { |
| for (String rel : autoTerminatedRelationships) { |
| addTextElement(element, "autoTerminatedRelationship", rel); |
| } |
| } |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add a Processor", e); |
| } |
| } |
| |
| |
| protected static void addFunnel(final Element parentElement, FunnelSchema funnelSchema) { |
| Document document = parentElement.getOwnerDocument(); |
| Element element = document.createElement("funnel"); |
| parentElement.appendChild(element); |
| |
| addTextElement(element, "id", funnelSchema.getId()); |
| |
| addPosition(element); |
| } |
| |
| protected static void addReportingTask(final Element parentElement, ReportingSchema reportingSchema) throws ConfigurationChangeException { |
| try { |
| final Document doc = parentElement.getOwnerDocument(); |
| final Element element = doc.createElement("reportingTask"); |
| parentElement.appendChild(element); |
| |
| addTextElement(element, "id", reportingSchema.getId()); |
| addTextElement(element, "name", reportingSchema.getName()); |
| addTextElement(element, "comment", reportingSchema.getComment()); |
| addTextElement(element, "class", reportingSchema.getReportingClass()); |
| addTextElement(element, "schedulingPeriod", reportingSchema.getSchedulingPeriod()); |
| addTextElement(element, "scheduledState", "RUNNING"); |
| addTextElement(element, "schedulingStrategy", reportingSchema.getSchedulingStrategy()); |
| |
| addConfiguration(element, reportingSchema.getProperties()); |
| |
| parentElement.appendChild(element); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the Provenance Reporting Task", e); |
| } |
| } |
| |
| protected static void addConfiguration(final Element element, Map<String, Object> elementConfig) { |
| final Document doc = element.getOwnerDocument(); |
| if (elementConfig == null) { |
| return; |
| } |
| for (final Map.Entry<String, Object> entry : elementConfig.entrySet()) { |
| |
| final Element propElement = doc.createElement("property"); |
| addTextElement(propElement, "name", entry.getKey()); |
| if (entry.getValue() != null) { |
| addTextElement(propElement, "value", entry.getValue().toString()); |
| } |
| |
| element.appendChild(propElement); |
| } |
| } |
| |
| protected static void addStyle(final Element parentElement) { |
| final Element element = parentElement.getOwnerDocument().createElement("styles"); |
| parentElement.appendChild(element); |
| } |
| |
| protected static void addRemoteProcessGroup(final Element parentElement, RemoteProcessGroupSchema remoteProcessGroupProperties) throws ConfigurationChangeException { |
| try { |
| final Document doc = parentElement.getOwnerDocument(); |
| final Element element = doc.createElement("remoteProcessGroup"); |
| parentElement.appendChild(element); |
| addTextElement(element, "id", remoteProcessGroupProperties.getId()); |
| addTextElement(element, "name", remoteProcessGroupProperties.getName()); |
| addPosition(element); |
| addTextElement(element, "comment", remoteProcessGroupProperties.getComment()); |
| // In the case we have multiple urls, select the first |
| addTextElement(element, "url", Arrays.asList(remoteProcessGroupProperties.getUrls().split(",")).get(0)); |
| addTextElement(element, "urls", remoteProcessGroupProperties.getUrls()); |
| addTextElement(element, "timeout", remoteProcessGroupProperties.getTimeout()); |
| addTextElement(element, "yieldPeriod", remoteProcessGroupProperties.getYieldPeriod()); |
| addTextElement(element, "transmitting", "true"); |
| addTextElement(element, "transportProtocol", remoteProcessGroupProperties.getTransportProtocol()); |
| addTextElement(element, "proxyHost", remoteProcessGroupProperties.getProxyHost()); |
| if (remoteProcessGroupProperties.getProxyPort() != null) { |
| addTextElement(element, "proxyPort", Integer.toString(remoteProcessGroupProperties.getProxyPort())); |
| } |
| addTextElement(element, "proxyUser", remoteProcessGroupProperties.getProxyUser()); |
| if (!StringUtils.isEmpty(remoteProcessGroupProperties.getProxyPassword())) { |
| addTextElement(element, "proxyPassword", remoteProcessGroupProperties.getProxyPassword()); |
| } |
| |
| List<RemotePortSchema> remoteInputPorts = remoteProcessGroupProperties.getInputPorts(); |
| for (RemotePortSchema remoteInputPortSchema : remoteInputPorts) { |
| addRemoteGroupPort(element, remoteInputPortSchema, "inputPort"); |
| } |
| |
| List<RemotePortSchema> remoteOutputPorts = remoteProcessGroupProperties.getOutputPorts(); |
| for (RemotePortSchema remoteOutputPortSchema : remoteOutputPorts) { |
| addRemoteGroupPort(element, remoteOutputPortSchema, "outputPort"); |
| } |
| addTextElement(element, "networkInterface", remoteProcessGroupProperties.getLocalNetworkInterface()); |
| |
| parentElement.appendChild(element); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the Remote Process Group", e); |
| } |
| } |
| |
| protected static void addRemoteGroupPort(final Element parentElement, RemotePortSchema inputPort, String tagName) throws ConfigurationChangeException { |
| try { |
| final Document doc = parentElement.getOwnerDocument(); |
| final Element element = doc.createElement(tagName); |
| parentElement.appendChild(element); |
| addTextElement(element, "id", inputPort.getId()); |
| addTextElement(element, "name", inputPort.getName()); |
| addPosition(element); |
| addTextElement(element, "comments", inputPort.getComment()); |
| addTextElement(element, "scheduledState", "RUNNING"); |
| addTextElement(element, "maxConcurrentTasks", String.valueOf(inputPort.getMax_concurrent_tasks())); |
| addTextElement(element, "useCompression", String.valueOf(inputPort.getUseCompression())); |
| |
| parentElement.appendChild(element); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the input port of the Remote Process Group", e); |
| } |
| } |
| |
| protected static void addConnection(final Element parentElement, ConnectionSchema connectionProperties, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException { |
| try { |
| final Document doc = parentElement.getOwnerDocument(); |
| final Element element = doc.createElement("connection"); |
| parentElement.appendChild(element); |
| |
| addTextElement(element, "id", connectionProperties.getId()); |
| addTextElement(element, "name", connectionProperties.getName()); |
| |
| final Element bendPointsElement = doc.createElement("bendPoints"); |
| element.appendChild(bendPointsElement); |
| |
| addTextElement(element, "labelIndex", "1"); |
| addTextElement(element, "zIndex", "0"); |
| |
| addConnectionSourceOrDestination(element, "source", connectionProperties.getSourceId(), parentGroupIdResolver); |
| addConnectionSourceOrDestination(element, "destination", connectionProperties.getDestinationId(), parentGroupIdResolver); |
| |
| List<String> sourceRelationshipNames = connectionProperties.getSourceRelationshipNames(); |
| if (sourceRelationshipNames.isEmpty()) { |
| addTextElement(element, "relationship", null); |
| } else { |
| for (String relationshipName : sourceRelationshipNames) { |
| addTextElement(element, "relationship", relationshipName); |
| } |
| } |
| |
| addTextElement(element, "maxWorkQueueSize", String.valueOf(connectionProperties.getMaxWorkQueueSize())); |
| addTextElement(element, "maxWorkQueueDataSize", connectionProperties.getMaxWorkQueueDataSize()); |
| |
| addTextElement(element, "flowFileExpiration", connectionProperties.getFlowfileExpiration()); |
| addTextElementIfNotNullOrEmpty(element, "queuePrioritizerClass", connectionProperties.getQueuePrioritizerClass()); |
| |
| parentElement.appendChild(element); |
| } catch (Exception e) { |
| throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the connection from the Processor to the input port of the Remote Process Group", e); |
| } |
| } |
| |
| protected static void addConnectionSourceOrDestination(Element element, String sourceOrDestination, String id, ParentGroupIdResolver parentGroupIdResolver) { |
| String idTag = sourceOrDestination + "Id"; |
| String groupIdTag = sourceOrDestination + "GroupId"; |
| String typeTag = sourceOrDestination + "Type"; |
| |
| String parentId; |
| String type; |
| |
| if ((parentId = parentGroupIdResolver.getRemoteInputPortParentId(id)) != null) { |
| type = "REMOTE_INPUT_PORT"; |
| } else if ((parentId = parentGroupIdResolver.getRemoteOutputPortParentId(id)) != null) { |
| type = "REMOTE_OUTPUT_PORT"; |
| } else if ((parentId = parentGroupIdResolver.getInputPortParentId(id)) != null) { |
| type = "INPUT_PORT"; |
| } else if ((parentId = parentGroupIdResolver.getOutputPortParentId(id)) != null) { |
| type = "OUTPUT_PORT"; |
| } else if ((parentId = parentGroupIdResolver.getFunnelParentId(id)) != null) { |
| type = "FUNNEL"; |
| } else { |
| parentId = parentGroupIdResolver.getProcessorParentId(id); |
| type = "PROCESSOR"; |
| } |
| |
| addTextElement(element, idTag, id); |
| if (parentId != null) { |
| addTextElement(element, groupIdTag, parentId); |
| } |
| addTextElement(element, typeTag, type); |
| } |
| |
| protected static void addPosition(final Element parentElement) { |
| final Element element = parentElement.getOwnerDocument().createElement("position"); |
| element.setAttribute("x", String.valueOf("0")); |
| element.setAttribute("y", String.valueOf("0")); |
| parentElement.appendChild(element); |
| } |
| |
| protected static void addTextElementIfNotNullOrEmpty(final Element element, final String name, final String value) { |
| StringUtil.doIfNotNullOrEmpty(value, s -> addTextElement(element, name, value)); |
| } |
| |
| protected static void addTextElement(final Element element, final String name, final String value) { |
| final Document doc = element.getOwnerDocument(); |
| final Element toAdd = doc.createElement(name); |
| toAdd.setTextContent(value); |
| element.appendChild(toAdd); |
| } |
| |
| public static final String PROPERTIES_FILE_APACHE_2_0_LICENSE = |
| " Licensed to the Apache Software Foundation (ASF) under one or more\n" + |
| "# contributor license agreements. See the NOTICE file distributed with\n" + |
| "# this work for additional information regarding copyright ownership.\n" + |
| "# The ASF licenses this file to You under the Apache License, Version 2.0\n" + |
| "# (the \"License\"); you may not use this file except in compliance with\n" + |
| "# the License. You may obtain a copy of the License at\n" + |
| "#\n" + |
| "# http://www.apache.org/licenses/LICENSE-2.0\n" + |
| "#\n" + |
| "# Unless required by applicable law or agreed to in writing, software\n" + |
| "# distributed under the License is distributed on an \"AS IS\" BASIS,\n" + |
| "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" + |
| "# See the License for the specific language governing permissions and\n" + |
| "# limitations under the License.\n" + |
| "\n"; |
| |
| } |
