| /* |
| * 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.knox.gateway.services.topology.impl; |
| |
| import org.apache.commons.digester3.Digester; |
| import org.apache.commons.digester3.binder.DigesterLoader; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.commons.io.FilenameUtils; |
| import org.apache.commons.io.monitor.FileAlterationListener; |
| import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; |
| import org.apache.commons.io.monitor.FileAlterationMonitor; |
| import org.apache.commons.io.monitor.FileAlterationObserver; |
| import org.apache.knox.gateway.GatewayMessages; |
| import org.apache.knox.gateway.GatewayServer; |
| import org.apache.knox.gateway.audit.api.Action; |
| import org.apache.knox.gateway.audit.api.ActionOutcome; |
| import org.apache.knox.gateway.audit.api.AuditServiceFactory; |
| import org.apache.knox.gateway.audit.api.Auditor; |
| import org.apache.knox.gateway.audit.api.ResourceType; |
| import org.apache.knox.gateway.audit.log4j.audit.AuditConstants; |
| import org.apache.knox.gateway.config.GatewayConfig; |
| import org.apache.knox.gateway.i18n.messages.MessagesFactory; |
| import org.apache.knox.gateway.service.definition.ServiceDefinition; |
| import org.apache.knox.gateway.service.definition.ServiceDefinitionChangeListener; |
| import org.apache.knox.gateway.services.ServiceType; |
| import org.apache.knox.gateway.services.ServiceLifecycleException; |
| import org.apache.knox.gateway.services.config.client.RemoteConfigurationRegistryClient; |
| import org.apache.knox.gateway.services.security.AliasService; |
| import org.apache.knox.gateway.services.topology.TopologyService; |
| import org.apache.knox.gateway.topology.ClusterConfigurationMonitorService; |
| import org.apache.knox.gateway.topology.Service; |
| import org.apache.knox.gateway.topology.Topology; |
| import org.apache.knox.gateway.topology.TopologyEvent; |
| import org.apache.knox.gateway.topology.TopologyListener; |
| import org.apache.knox.gateway.topology.TopologyMonitor; |
| import org.apache.knox.gateway.topology.TopologyProvider; |
| import org.apache.knox.gateway.topology.Version; |
| import org.apache.knox.gateway.topology.builder.TopologyBuilder; |
| import org.apache.knox.gateway.topology.discovery.ClusterConfigurationMonitor; |
| import org.apache.knox.gateway.topology.discovery.ServiceDiscovery; |
| import org.apache.knox.gateway.topology.monitor.RemoteConfigurationMonitor; |
| import org.apache.knox.gateway.topology.monitor.RemoteConfigurationMonitorFactory; |
| import org.apache.knox.gateway.topology.simple.ProviderConfigurationParser; |
| import org.apache.knox.gateway.topology.simple.SimpleDescriptor; |
| import org.apache.knox.gateway.topology.simple.SimpleDescriptorFactory; |
| import org.apache.knox.gateway.topology.simple.SimpleDescriptorHandler; |
| import org.apache.knox.gateway.topology.validation.TopologyValidator; |
| import org.apache.knox.gateway.topology.xml.AmbariFormatXmlTopologyRules; |
| import org.apache.knox.gateway.topology.xml.KnoxFormatXmlTopologyRules; |
| import org.apache.knox.gateway.util.ServiceDefinitionsLoader; |
| import org.eclipse.persistence.jaxb.JAXBContextProperties; |
| import org.xml.sax.SAXException; |
| |
| import javax.xml.bind.JAXBContext; |
| import javax.xml.bind.JAXBException; |
| import javax.xml.bind.Marshaller; |
| import java.io.File; |
| import java.io.FileFilter; |
| import java.io.IOException; |
| import java.nio.charset.StandardCharsets; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| |
| import static org.apache.commons.digester3.binder.DigesterLoader.newLoader; |
| |
| public class DefaultTopologyService extends FileAlterationListenerAdaptor implements TopologyService, TopologyMonitor, |
| TopologyProvider, FileFilter, FileAlterationListener, ServiceDefinitionChangeListener { |
| |
| private static final JAXBContext jaxbContext = getJAXBContext(); |
| |
| private static Auditor auditor = AuditServiceFactory.getAuditService().getAuditor( |
| AuditConstants.DEFAULT_AUDITOR_NAME, AuditConstants.KNOX_SERVICE_NAME, |
| AuditConstants.KNOX_COMPONENT_NAME); |
| |
| private static final List<String> SUPPORTED_TOPOLOGY_FILE_EXTENSIONS = Arrays.asList("xml", "conf"); |
| |
| private static GatewayMessages log = MessagesFactory.get(GatewayMessages.class); |
| private static DigesterLoader digesterLoader = newLoader(new KnoxFormatXmlTopologyRules(), |
| new AmbariFormatXmlTopologyRules()); |
| private List<FileAlterationMonitor> monitors = new ArrayList<>(); |
| private File topologiesDirectory; |
| private File sharedProvidersDirectory; |
| private File descriptorsDirectory; |
| |
| private DescriptorsMonitor descriptorsMonitor; |
| |
| private Set<TopologyListener> listeners; |
| private Map<File, Topology> topologies; |
| private AliasService aliasService; |
| |
| private RemoteConfigurationMonitor remoteMonitor; |
| |
| private GatewayConfig config; |
| |
| private static JAXBContext getJAXBContext() { |
| String pkgName = Topology.class.getPackage().getName(); |
| String bindingFile = pkgName.replace(".", "/") + "/topology_binding-xml.xml"; |
| |
| Map<String, Object> properties = new HashMap<>(1); |
| properties.put(JAXBContextProperties.OXM_METADATA_SOURCE, bindingFile); |
| try { |
| return JAXBContext.newInstance(pkgName, Topology.class.getClassLoader(), properties); |
| } catch (JAXBException e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| private Topology loadTopology(File file) throws IOException, SAXException, InterruptedException { |
| final long TIMEOUT = 250; //ms |
| final long DELAY = 50; //ms |
| log.loadingTopologyFile(file.getAbsolutePath()); |
| Topology topology; |
| long start = System.currentTimeMillis(); |
| while (true) { |
| try { |
| topology = loadTopologyAttempt(file); |
| break; |
| } catch (IOException | SAXException e) { |
| if (System.currentTimeMillis() - start < TIMEOUT) { |
| log.failedToLoadTopologyRetrying(file.getAbsolutePath(), Long.toString(DELAY), e); |
| Thread.sleep(DELAY); |
| } else { |
| throw e; |
| } |
| } |
| } |
| return topology; |
| } |
| |
| private Topology loadTopologyAttempt(File file) throws IOException, SAXException { |
| Topology topology; |
| Digester digester = digesterLoader.newDigester(); |
| TopologyBuilder topologyBuilder = digester.parse(FileUtils.openInputStream(file)); |
| if (null == topologyBuilder) { |
| return null; |
| } |
| topology = topologyBuilder.build(); |
| topology.setUri(file.toURI()); |
| topology.setName(FilenameUtils.removeExtension(file.getName())); |
| topology.setTimestamp(file.lastModified()); |
| return topology; |
| } |
| |
| private void redeployTopology(Topology topology) { |
| File topologyFile = new File(topology.getUri()); |
| try { |
| TopologyValidator tv = new TopologyValidator(topology); |
| |
| if(!tv.validateTopology()) { |
| if(config != null && config.isTopologyValidationEnabled()) { |
| /* If strict validation enabled we fail */ |
| throw new SAXException(tv.getErrorString()); |
| } else { |
| /* Log and move on */ |
| log.failedToValidateTopology(topology.getName(), tv.getErrorString()); |
| } |
| } |
| |
| long start = System.currentTimeMillis(); |
| long limit = 1000L; // One second. |
| long elapsed = 1; |
| while (elapsed <= limit) { |
| try { |
| long origTimestamp = topologyFile.lastModified(); |
| long setTimestamp = Math.max(System.currentTimeMillis(), topologyFile.lastModified() + elapsed); |
| if(topologyFile.setLastModified(setTimestamp)) { |
| long newTimstamp = topologyFile.lastModified(); |
| if(newTimstamp > origTimestamp) { |
| break; |
| } else { |
| Thread.sleep(10); |
| elapsed = System.currentTimeMillis() - start; |
| continue; |
| } |
| } else { |
| auditor.audit(Action.REDEPLOY, topology.getName(), ResourceType.TOPOLOGY, |
| ActionOutcome.FAILURE); |
| log.failedToRedeployTopology(topology.getName()); |
| break; |
| } |
| } catch (InterruptedException e) { |
| auditor.audit(Action.REDEPLOY, topology.getName(), ResourceType.TOPOLOGY, ActionOutcome.FAILURE); |
| log.failedToRedeployTopology(topology.getName(), e); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } catch (SAXException e) { |
| auditor.audit(Action.REDEPLOY, topology.getName(), ResourceType.TOPOLOGY, ActionOutcome.FAILURE); |
| log.failedToRedeployTopology(topology.getName(), e); |
| } |
| } |
| |
| private List<TopologyEvent> createChangeEvents( |
| Map<File, Topology> oldTopologies, |
| Map<File, Topology> newTopologies) { |
| ArrayList<TopologyEvent> events = new ArrayList<>(); |
| // Go through the old topologies and find anything that was deleted. |
| for (Entry<File, Topology> oldTopology : oldTopologies.entrySet()) { |
| if (!newTopologies.containsKey(oldTopology.getKey())) { |
| events.add(new TopologyEvent(TopologyEvent.Type.DELETED, oldTopology.getValue())); |
| } |
| } |
| // Go through the new topologies and figure out what was updated vs added. |
| for (Entry<File, Topology> newTopology : newTopologies.entrySet()) { |
| if (oldTopologies.containsKey(newTopology.getKey())) { |
| Topology oldTopology = oldTopologies.get(newTopology.getKey()); |
| if (newTopology.getValue().getTimestamp() > oldTopology.getTimestamp()) { |
| events.add(new TopologyEvent(TopologyEvent.Type.UPDATED, newTopology.getValue())); |
| } |
| } else { |
| events.add(new TopologyEvent(TopologyEvent.Type.CREATED, newTopology.getValue())); |
| } |
| } |
| return events; |
| } |
| |
| private File calculateAbsoluteTopologiesDir(GatewayConfig config) { |
| File topoDir = new File(config.getGatewayTopologyDir()); |
| topoDir = topoDir.getAbsoluteFile(); |
| return topoDir; |
| } |
| |
| private File calculateAbsoluteConfigDir(GatewayConfig config) { |
| File configDir; |
| |
| String path = config.getGatewayConfDir(); |
| configDir = (path != null) ? new File(path) : (new File(config.getGatewayTopologyDir())).getParentFile(); |
| |
| return configDir.getAbsoluteFile(); |
| } |
| |
| private void initListener(FileAlterationMonitor monitor, |
| File directory, |
| FileFilter filter, |
| FileAlterationListener listener) { |
| monitors.add(monitor); |
| FileAlterationObserver observer = new FileAlterationObserver(directory, filter); |
| observer.addListener(listener); |
| monitor.addObserver(observer); |
| } |
| |
| private void initListener(File directory, FileFilter filter, FileAlterationListener listener) { |
| // Increasing the monitoring interval to 5 seconds as profiling has shown |
| // this is rather expensive in terms of generated garbage objects. |
| initListener(new FileAlterationMonitor(5000L), directory, filter, listener); |
| } |
| |
| private Map<File, Topology> loadTopologies(File directory) { |
| Map<File, Topology> map = new HashMap<>(); |
| if (directory.isDirectory() && directory.canRead()) { |
| File[] existingTopologies = directory.listFiles(this); |
| if (existingTopologies != null) { |
| for (File file : existingTopologies) { |
| try { |
| Topology loadTopology = loadTopology(file); |
| if (null != loadTopology) { |
| map.put(file, loadTopology); |
| } else { |
| auditor.audit(Action.LOAD, file.getAbsolutePath(), ResourceType.TOPOLOGY, |
| ActionOutcome.FAILURE); |
| log.failedToLoadTopology(file.getAbsolutePath()); |
| } |
| } catch (Exception e) { |
| // Maybe it makes sense to throw exception |
| auditor.audit(Action.LOAD, file.getAbsolutePath(), ResourceType.TOPOLOGY, |
| ActionOutcome.FAILURE); |
| log.failedToLoadTopology(file.getAbsolutePath(), e); |
| } |
| } |
| } |
| } |
| return map; |
| } |
| |
| public void setAliasService(AliasService as) { |
| this.aliasService = as; |
| } |
| |
| @Override |
| public void deployTopology(Topology t){ |
| |
| try { |
| File temp = new File(topologiesDirectory.getAbsolutePath() + "/" + t.getName() + ".xml.temp"); |
| Marshaller mr = jaxbContext.createMarshaller(); |
| |
| mr.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true); |
| mr.marshal(t, temp); |
| |
| File topology = new File(topologiesDirectory.getAbsolutePath() + "/" + t.getName() + ".xml"); |
| if(!temp.renameTo(topology)) { |
| FileUtils.forceDelete(temp); |
| throw new IOException("Could not rename temp file"); |
| } |
| |
| // This code will check if the topology is valid, and retrieve the errors if it is not. |
| TopologyValidator validator = new TopologyValidator( topology.getAbsolutePath() ); |
| if( !validator.validateTopology() ){ |
| throw new SAXException( validator.getErrorString() ); |
| } |
| |
| |
| } catch (JAXBException | SAXException | IOException e) { |
| auditor.audit(Action.DEPLOY, t.getName(), ResourceType.TOPOLOGY, ActionOutcome.FAILURE); |
| log.failedToDeployTopology(t.getName(), e); |
| } |
| reloadTopologies(); |
| } |
| |
| @Override |
| public void redeployTopologies(String topologyName) { |
| |
| for (Topology topology : getTopologies()) { |
| if (topologyName == null || topologyName.equals(topology.getName())) { |
| redeployTopology(topology); |
| } |
| } |
| |
| } |
| |
| @Override |
| public void reloadTopologies() { |
| try { |
| synchronized (this) { |
| Map<File, Topology> oldTopologies = topologies; |
| Map<File, Topology> newTopologies = loadTopologies(topologiesDirectory); |
| List<TopologyEvent> events = createChangeEvents(oldTopologies, newTopologies); |
| topologies = newTopologies; |
| notifyChangeListeners(events); |
| } |
| } catch (Exception e) { |
| // Maybe it makes sense to throw exception |
| log.failedToReloadTopologies(e); |
| } |
| } |
| |
| @Override |
| public void deleteTopology(Topology t) { |
| File topoDir = topologiesDirectory; |
| |
| if(topoDir.isDirectory() && topoDir.canRead()) { |
| for (File f : listFiles(topoDir)) { |
| String fName = FilenameUtils.getBaseName(f.getName()); |
| if(fName.equals(t.getName())) { |
| f.delete(); |
| } |
| } |
| } |
| reloadTopologies(); |
| } |
| |
| private void notifyChangeListeners(List<TopologyEvent> events) { |
| for (TopologyListener listener : listeners) { |
| try { |
| listener.handleTopologyEvent(events); |
| } catch (RuntimeException e) { |
| auditor.audit(Action.LOAD, "Topology_Event", ResourceType.TOPOLOGY, ActionOutcome.FAILURE); |
| log.failedToHandleTopologyEvents(e); |
| } |
| } |
| } |
| |
| @Override |
| public Map<String, List<String>> getServiceTestURLs(Topology t, GatewayConfig config) { |
| File tFile = null; |
| Map<String, List<String>> urls = new HashMap<>(); |
| if (topologiesDirectory.isDirectory() && topologiesDirectory.canRead()) { |
| for (File f : listFiles(topologiesDirectory)) { |
| if (FilenameUtils.removeExtension(f.getName()).equals(t.getName())) { |
| tFile = f; |
| } |
| } |
| } |
| Set<ServiceDefinition> defs; |
| if(tFile != null) { |
| defs = ServiceDefinitionsLoader.getServiceDefinitions(new File(config.getGatewayServicesDir())); |
| |
| for(ServiceDefinition def : defs) { |
| urls.put(def.getRole(), def.getTestURLs()); |
| } |
| } |
| return urls; |
| } |
| |
| @Override |
| public Collection<Topology> getTopologies() { |
| Map<File, Topology> map = topologies; |
| return Collections.unmodifiableCollection(map.values()); |
| } |
| |
| @Override |
| public boolean deployProviderConfiguration(String name, String content) { |
| boolean result; |
| |
| // Whether the remote configuration registry is being employed or not, write the file locally |
| result = writeConfig(sharedProvidersDirectory, name, content); |
| |
| // If the remote configuration registry is being employed, persist it there also |
| if (remoteMonitor != null) { |
| RemoteConfigurationRegistryClient client = remoteMonitor.getClient(); |
| if (client != null) { |
| String entryPath = "/knox/config/shared-providers/" + name; |
| client.createEntry(entryPath, content); |
| result = (client.getEntryData(entryPath) != null); |
| } |
| } |
| |
| return result; |
| } |
| |
| @Override |
| public Collection<File> getProviderConfigurations() { |
| List<File> providerConfigs = new ArrayList<>(); |
| for (File providerConfig : listFiles(sharedProvidersDirectory)) { |
| if (SharedProviderConfigMonitor.SUPPORTED_EXTENSIONS.contains(FilenameUtils.getExtension(providerConfig.getName()))) { |
| providerConfigs.add(providerConfig); |
| } |
| } |
| return providerConfigs; |
| } |
| |
| @Override |
| public boolean deleteProviderConfiguration(String name) { |
| return deleteProviderConfiguration(name, false); |
| } |
| |
| @Override |
| public boolean deleteProviderConfiguration(String name, boolean force) { |
| boolean result = false; |
| |
| // Determine if the file exists, and if so, if there are any descriptors referencing it |
| boolean hasReferences = false; |
| File providerConfig = getExistingFile(sharedProvidersDirectory, name); |
| if (providerConfig != null) { |
| List<String> references = descriptorsMonitor.getReferencingDescriptors(providerConfig.getAbsolutePath()); |
| hasReferences = !references.isEmpty(); |
| } else { |
| result = true; // If it already does NOT exist, then the delete effectively succeeded |
| } |
| |
| // If the local file does not exist, or it does exist and there are NOT any referencing descriptors |
| if (force || (providerConfig == null || !hasReferences)) { |
| |
| // If the remote config monitor is configured, attempt to delete the provider configuration from the remote |
| // registry, even if it does not exist locally. |
| deleteRemoteEntry("/knox/config/shared-providers", name); |
| |
| // Whether the remote configuration registry is being employed or not, delete the local file if it exists |
| result = providerConfig == null || !providerConfig.exists() || providerConfig.delete(); |
| |
| } else { |
| log.preventedDeletionOfSharedProviderConfiguration(providerConfig.getAbsolutePath()); |
| } |
| |
| return result; |
| } |
| |
| @Override |
| public boolean deployDescriptor(String name, String content) { |
| boolean result; |
| |
| // Whether the remote configuration registry is being employed or not, write the file locally |
| result = writeConfig(descriptorsDirectory, name, content); |
| |
| // If the remote configuration registry is being employed, persist it there also |
| if (remoteMonitor != null) { |
| RemoteConfigurationRegistryClient client = remoteMonitor.getClient(); |
| if (client != null) { |
| String entryPath = "/knox/config/descriptors/" + name; |
| client.createEntry(entryPath, content); |
| result = (client.getEntryData(entryPath) != null); |
| } |
| } |
| |
| return result; |
| } |
| |
| @Override |
| public Collection<File> getDescriptors() { |
| List<File> descriptors = new ArrayList<>(); |
| for (File descriptor : listFiles(descriptorsDirectory)) { |
| if (DescriptorsMonitor.SUPPORTED_EXTENSIONS.contains(FilenameUtils.getExtension(descriptor.getName()))) { |
| descriptors.add(descriptor); |
| } |
| } |
| return descriptors; |
| } |
| |
| @Override |
| public boolean deleteDescriptor(String name) { |
| boolean result; |
| |
| // If the remote config monitor is configured, delete the descriptor from the remote registry |
| deleteRemoteEntry("/knox/config/descriptors", name); |
| |
| // Whether the remote configuration registry is being employed or not, delete the local file |
| File descriptor = getExistingFile(descriptorsDirectory, name); |
| result = (descriptor == null) || descriptor.delete(); |
| |
| return result; |
| } |
| |
| @Override |
| public void addTopologyChangeListener(TopologyListener listener) { |
| listeners.add(listener); |
| } |
| |
| @Override |
| public void onServiceDefinitionChange(String name, String role, String version) { |
| getTopologies().stream().filter(topology -> topology.getServices().stream().anyMatch(service -> isRelevantService(service, role, name, version))).forEach(topology -> { |
| log.redeployingTopologyOnServiceDefinitionChange(topology.getName(), name, role, version); |
| redeployTopology(topology); |
| }); |
| } |
| |
| private boolean isRelevantService(Service service, String role, String name, String version) { |
| return service.getRole().equalsIgnoreCase(role) |
| && (service.getName() == null || service.getName().equalsIgnoreCase(name) && (service.getVersion() == null || service.getVersion().equals(new Version(version)))); |
| } |
| |
| @Override |
| public void startMonitor() throws Exception { |
| // Start the local configuration monitors |
| for (FileAlterationMonitor monitor : monitors) { |
| monitor.start(); |
| } |
| |
| // Start the remote configuration monitor, if it has been initialized |
| if (remoteMonitor != null) { |
| try { |
| remoteMonitor.start(); |
| } catch (Exception e) { |
| log.remoteConfigurationMonitorStartFailure(remoteMonitor.getClass().getTypeName(), e.getLocalizedMessage()); |
| } |
| } |
| } |
| |
| @Override |
| public void stopMonitor() throws Exception { |
| // Stop the local configuration monitors |
| for (FileAlterationMonitor monitor : monitors) { |
| monitor.stop(); |
| } |
| |
| // Stop the remote configuration monitor, if it has been initialized |
| if (remoteMonitor != null) { |
| remoteMonitor.stop(); |
| } |
| } |
| |
| @Override |
| public boolean accept(File file) { |
| boolean accept = false; |
| if (!file.isDirectory() && file.canRead()) { |
| String extension = FilenameUtils.getExtension(file.getName()); |
| if (SUPPORTED_TOPOLOGY_FILE_EXTENSIONS.contains(extension)) { |
| accept = true; |
| } |
| } |
| return accept; |
| } |
| |
| @Override |
| public void onFileCreate(File file) { |
| onFileChange(file); |
| } |
| |
| @Override |
| public void onFileDelete(java.io.File file) { |
| onFileChange(file); |
| } |
| |
| @Override |
| public void onFileChange(File file) { |
| reloadTopologies(); |
| } |
| |
| @Override |
| public void stop() { |
| |
| } |
| |
| @Override |
| public void start() { |
| // Register a cluster configuration monitor listener for change notifications |
| ClusterConfigurationMonitorService ccms = |
| GatewayServer.getGatewayServices().getService(ServiceType.CLUSTER_CONFIGURATION_MONITOR_SERVICE); |
| ccms.addListener(new TopologyDiscoveryTrigger(this, ccms)); |
| } |
| |
| @Override |
| public void init(GatewayConfig config, Map<String, String> options) throws ServiceLifecycleException { |
| |
| this.config = config; |
| String gatewayConfDir = config.getGatewayConfDir(); |
| if (gatewayConfDir != null) { |
| System.setProperty(ServiceDiscovery.CONFIG_DIR_PROPERTY, gatewayConfDir); |
| } |
| |
| try { |
| listeners = new HashSet<>(); |
| topologies = new HashMap<>(); |
| |
| topologiesDirectory = calculateAbsoluteTopologiesDir(config); |
| |
| File configDirectory = calculateAbsoluteConfigDir(config); |
| descriptorsDirectory = new File(configDirectory, "descriptors"); |
| sharedProvidersDirectory = new File(configDirectory, "shared-providers"); |
| |
| // Add support for conf/topologies |
| initListener(topologiesDirectory, this, this); |
| |
| // Add support for conf/descriptors |
| descriptorsMonitor = new DescriptorsMonitor(config, topologiesDirectory, aliasService); |
| initListener(descriptorsDirectory, |
| descriptorsMonitor, |
| descriptorsMonitor); |
| log.monitoringDescriptorChangesInDirectory(descriptorsDirectory.getAbsolutePath()); |
| |
| // Add support for conf/shared-providers |
| SharedProviderConfigMonitor spm = new SharedProviderConfigMonitor(descriptorsMonitor, descriptorsDirectory); |
| initListener(sharedProvidersDirectory, spm, spm); |
| log.monitoringProviderConfigChangesInDirectory(sharedProvidersDirectory.getAbsolutePath()); |
| |
| // For all the descriptors currently in the descriptors dir at start-up time, determine if topology regeneration |
| // is required. |
| // This happens prior to the start-up loading of the topologies. |
| String[] descriptorFilenames = descriptorsDirectory.list(); |
| if (descriptorFilenames != null) { |
| for (String descriptorFilename : descriptorFilenames) { |
| if (DescriptorsMonitor.isDescriptorFile(descriptorFilename)) { |
| String topologyName = FilenameUtils.getBaseName(descriptorFilename); |
| File existingDescriptorFile = getExistingFile(descriptorsDirectory, topologyName); |
| |
| // If there isn't a corresponding topology file, or if the descriptor has been modified since the |
| // corresponding topology file was generated, then trigger generation of one |
| File matchingTopologyFile = getExistingFile(topologiesDirectory, topologyName); |
| if (matchingTopologyFile == null || matchingTopologyFile.lastModified() < existingDescriptorFile.lastModified()) { |
| descriptorsMonitor.onFileChange(existingDescriptorFile); |
| } else { |
| // If regeneration is NOT required, then we at least need to report the provider configuration |
| // reference relationship (KNOX-1144) |
| String normalizedDescriptorPath = FilenameUtils.normalize(existingDescriptorFile.getAbsolutePath()); |
| |
| // Parse the descriptor to determine the provider config reference |
| SimpleDescriptor sd = SimpleDescriptorFactory.parse(normalizedDescriptorPath); |
| if (sd != null) { |
| File referencedProviderConfig = |
| getExistingFile(sharedProvidersDirectory, FilenameUtils.getBaseName(sd.getProviderConfig())); |
| if (referencedProviderConfig != null) { |
| List<String> references = |
| descriptorsMonitor.getReferencingDescriptors(referencedProviderConfig.getAbsolutePath()); |
| if (!references.contains(normalizedDescriptorPath)) { |
| references.add(normalizedDescriptorPath); |
| } |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| // Initialize the remote configuration monitor, if it has been configured |
| remoteMonitor = RemoteConfigurationMonitorFactory.get(config); |
| |
| } catch (IOException io) { |
| throw new ServiceLifecycleException(io.getMessage(), io); |
| } |
| } |
| |
| /** |
| * Delete the entry in the remote configuration registry, which matches the specified resource name. |
| * |
| * @param entryParent The remote registry path in which the entry exists. |
| * @param name The name of the entry (typically without any file extension). |
| * |
| * @return true, if the entry is deleted, or did not exist; otherwise, false. |
| */ |
| private boolean deleteRemoteEntry(String entryParent, String name) { |
| boolean result = true; |
| |
| if (remoteMonitor != null) { |
| RemoteConfigurationRegistryClient client = remoteMonitor.getClient(); |
| if (client != null) { |
| List<String> existingProviderConfigs = client.listChildEntries(entryParent); |
| for (String entryName : existingProviderConfigs) { |
| if (FilenameUtils.getBaseName(entryName).equals(name)) { |
| String entryPath = entryParent + "/" + entryName; |
| client.deleteEntry(entryPath); |
| result = !client.entryExists(entryPath); |
| if (!result) { |
| log.failedToDeletedRemoteConfigFile("descriptor", name); |
| } |
| break; |
| } |
| } |
| } |
| } |
| |
| return result; |
| } |
| |
| /** |
| * Utility method for listing the files in the specified directory. |
| * This method is "nicer" than the File#listFiles() because it will not return null. |
| * |
| * @param directory The directory whose files should be returned. |
| * |
| * @return A List of the Files on the directory. |
| */ |
| private static List<File> listFiles(File directory) { |
| List<File> result; |
| File[] files = directory.listFiles(); |
| if (files != null) { |
| result = Arrays.asList(files); |
| } else { |
| result = Collections.emptyList(); |
| } |
| return result; |
| } |
| |
| /** |
| * Search for a file in the specified directory whose base name (filename without extension) matches the |
| * specified basename. |
| * |
| * @param directory The directory in which to search. |
| * @param basename The basename of interest. |
| * |
| * @return The matching File |
| */ |
| private static File getExistingFile(File directory, String basename) { |
| File match = null; |
| for (File file : listFiles(directory)) { |
| if (FilenameUtils.getBaseName(file.getName()).equals(basename)) { |
| match = file; |
| break; |
| } |
| } |
| return match; |
| } |
| |
| /** |
| * Write the specified content to a file. |
| * |
| * @param dest The destination directory. |
| * @param name The name of the file. |
| * @param content The contents of the file. |
| * |
| * @return true, if the write succeeds; otherwise, false. |
| */ |
| private static boolean writeConfig(File dest, String name, String content) { |
| boolean result = false; |
| |
| File destFile = new File(dest, name); |
| try { |
| FileUtils.writeStringToFile(destFile, content, StandardCharsets.UTF_8); |
| log.wroteConfigurationFile(destFile.getAbsolutePath()); |
| result = true; |
| } catch (IOException e) { |
| log.failedToWriteConfigurationFile(destFile.getAbsolutePath(), e); |
| } |
| |
| return result; |
| } |
| |
| |
| /** |
| * Change handler for simple descriptors |
| */ |
| public static class DescriptorsMonitor extends FileAlterationListenerAdaptor |
| implements FileFilter { |
| |
| static final List<String> SUPPORTED_EXTENSIONS = new ArrayList<>(); |
| static { |
| SUPPORTED_EXTENSIONS.add("json"); |
| SUPPORTED_EXTENSIONS.add("yml"); |
| SUPPORTED_EXTENSIONS.add("yaml"); |
| } |
| |
| private GatewayConfig gatewayConfig; |
| |
| private File topologiesDir; |
| |
| private AliasService aliasService; |
| |
| private Map<String, List<String>> providerConfigReferences = new HashMap<>(); |
| |
| |
| static boolean isDescriptorFile(String filename) { |
| return SUPPORTED_EXTENSIONS.contains(FilenameUtils.getExtension(filename)); |
| } |
| |
| public DescriptorsMonitor(GatewayConfig config, File topologiesDir, AliasService aliasService) { |
| this.gatewayConfig = config; |
| this.topologiesDir = topologiesDir; |
| this.aliasService = aliasService; |
| } |
| |
| List<String> getReferencingDescriptors(String providerConfigPath) { |
| String normalizedPath = FilenameUtils.normalize(providerConfigPath); |
| return providerConfigReferences.computeIfAbsent(normalizedPath, p -> new ArrayList<>()); |
| } |
| |
| @Override |
| public void onFileCreate(File file) { |
| onFileChange(file); |
| } |
| |
| @Override |
| public void onFileDelete(File file) { |
| // For simple descriptors, we need to make sure to delete any corresponding full topology descriptors to trigger undeployment |
| for (String ext : DefaultTopologyService.SUPPORTED_TOPOLOGY_FILE_EXTENSIONS) { |
| File topologyFile = |
| new File(topologiesDir, FilenameUtils.getBaseName(file.getName()) + "." + ext); |
| if (topologyFile.exists()) { |
| log.deletingTopologyForDescriptorDeletion(topologyFile.getName(), file.getName()); |
| topologyFile.delete(); |
| } |
| } |
| |
| String normalizedFilePath = FilenameUtils.normalize(file.getAbsolutePath()); |
| String reference = null; |
| for (Map.Entry<String, List<String>> entry : providerConfigReferences.entrySet()) { |
| if (entry.getValue().contains(normalizedFilePath)) { |
| reference = entry.getKey(); |
| break; |
| } |
| } |
| |
| if (reference != null) { |
| providerConfigReferences.get(reference).remove(normalizedFilePath); |
| log.removedProviderConfigurationReference(normalizedFilePath, reference); |
| } |
| } |
| |
| @Override |
| public void onFileChange(File file) { |
| try { |
| // When a simple descriptor has been created or modified, generate the new topology descriptor |
| Map<String, File> result = SimpleDescriptorHandler.handle(gatewayConfig, file, topologiesDir, aliasService); |
| log.generatedTopologyForDescriptorChange(result.get(SimpleDescriptorHandler.RESULT_TOPOLOGY).getName(), |
| file.getName()); |
| |
| // Add the provider config reference relationship for handling updates to the provider config |
| String providerConfig = |
| FilenameUtils.normalize(result.get(SimpleDescriptorHandler.RESULT_REFERENCE).getAbsolutePath()); |
| if (!providerConfigReferences.containsKey(providerConfig)) { |
| providerConfigReferences.put(providerConfig, new ArrayList<>()); |
| } |
| List<String> refs = providerConfigReferences.get(providerConfig); |
| String descriptorName = FilenameUtils.normalize(file.getAbsolutePath()); |
| if (!refs.contains(descriptorName)) { |
| // Need to check if descriptor had previously referenced another provider config, so it can be removed |
| for (List<String> descs : providerConfigReferences.values()) { |
| descs.remove(descriptorName); |
| } |
| |
| // Add the current reference relationship |
| refs.add(descriptorName); |
| log.addedProviderConfigurationReference(descriptorName, providerConfig); |
| } |
| } catch (IllegalArgumentException e) { |
| log.simpleDescriptorHandlingError(file.getName(), e); |
| |
| // If the referenced provider configuration is invalid, remove any existing reference relationships for the |
| // referencing descriptor. |
| String descriptorName = FilenameUtils.normalize(file.getAbsolutePath()); |
| // Need to check if descriptor had previously referenced another provider config, so it can be removed |
| for (List<String> descs : providerConfigReferences.values()) { |
| descs.remove(descriptorName); |
| } |
| } catch (Exception e) { |
| log.simpleDescriptorHandlingError(file.getName(), e); |
| } |
| } |
| |
| @Override |
| public boolean accept(File file) { |
| boolean accept = false; |
| if (!file.isDirectory() && file.canRead()) { |
| String extension = FilenameUtils.getExtension(file.getName()); |
| if (SUPPORTED_EXTENSIONS.contains(extension)) { |
| accept = true; |
| } |
| } |
| return accept; |
| } |
| } |
| |
| /** |
| * Change handler for shared provider configurations |
| */ |
| public static class SharedProviderConfigMonitor extends FileAlterationListenerAdaptor implements FileFilter { |
| |
| static final List<String> SUPPORTED_EXTENSIONS = ProviderConfigurationParser.SUPPORTED_EXTENSIONS; |
| |
| private DescriptorsMonitor descriptorsMonitor; |
| private File descriptorsDir; |
| |
| |
| SharedProviderConfigMonitor(DescriptorsMonitor descMonitor, File descriptorsDir) { |
| this.descriptorsMonitor = descMonitor; |
| this.descriptorsDir = descriptorsDir; |
| } |
| |
| @Override |
| public void onFileCreate(File file) { |
| onFileChange(file); |
| } |
| |
| @Override |
| public void onFileDelete(File file) { |
| onFileChange(file); |
| } |
| |
| @Override |
| public void onFileChange(File file) { |
| // For shared provider configuration, we need to update any simple descriptors that reference it |
| for (File descriptor : getReferencingDescriptors(file)) { |
| descriptor.setLastModified(System.currentTimeMillis()); |
| } |
| } |
| |
| private List<File> getReferencingDescriptors(File sharedProviderConfig) { |
| List<File> references = new ArrayList<>(); |
| |
| for (File descriptor : listFiles(descriptorsDir)) { |
| if (DescriptorsMonitor.SUPPORTED_EXTENSIONS.contains(FilenameUtils.getExtension(descriptor.getName()))) { |
| for (String reference : descriptorsMonitor.getReferencingDescriptors(FilenameUtils.normalize(sharedProviderConfig.getAbsolutePath()))) { |
| references.add(new File(reference)); |
| } |
| } |
| } |
| |
| return references; |
| } |
| |
| @Override |
| public boolean accept(File file) { |
| boolean accept = false; |
| if (!file.isDirectory() && file.canRead()) { |
| String extension = FilenameUtils.getExtension(file.getName()); |
| if (SUPPORTED_EXTENSIONS.contains(extension)) { |
| accept = true; |
| } |
| } |
| return accept; |
| } |
| } |
| |
| /** |
| * Listener for Ambari config change events, which will trigger re-generation (including re-discovery) of the |
| * affected topologies. |
| */ |
| private static class TopologyDiscoveryTrigger implements ClusterConfigurationMonitor.ConfigurationChangeListener { |
| |
| private TopologyService topologyService; |
| private ClusterConfigurationMonitorService ccms; |
| |
| TopologyDiscoveryTrigger(TopologyService topologyService, ClusterConfigurationMonitorService ccms) { |
| this.topologyService = topologyService; |
| this.ccms = ccms; |
| } |
| |
| @Override |
| public void onConfigurationChange(String source, String clusterName) { |
| log.noticedClusterConfigurationChange(source, clusterName); |
| try { |
| boolean affectedDescriptors = false; |
| // Identify any descriptors associated with the cluster configuration change |
| for (File descriptor : topologyService.getDescriptors()) { |
| String descriptorContent = FileUtils.readFileToString(descriptor, StandardCharsets.UTF_8); |
| if (descriptorContent.contains(source) && descriptorContent.contains(clusterName)) { |
| affectedDescriptors = true; |
| log.triggeringTopologyRegeneration(source, clusterName, descriptor.getAbsolutePath()); |
| // 'Touch' the descriptor to trigger re-generation of the associated topology |
| descriptor.setLastModified(System.currentTimeMillis()); |
| } |
| } |
| |
| if (!affectedDescriptors) { |
| // If not descriptors are affected by this configuration, then clear the cache to prevent future notifications |
| ccms.clearCache(source, clusterName); |
| } |
| } catch (Exception e) { |
| log.errorRespondingToConfigChange(source, clusterName, e); |
| } |
| } |
| } |
| |
| } |
