You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by nc...@apache.org on 2017/09/13 15:01:50 UTC
ambari git commit: AMBARI-21941. Stack selection page does not load
the HDP stacks [Intermittent] (ncole)
Repository: ambari
Updated Branches:
refs/heads/branch-2.6 536cfc390 -> 08db087b8
AMBARI-21941. Stack selection page does not load the HDP stacks [Intermittent] (ncole)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/08db087b
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/08db087b
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/08db087b
Branch: refs/heads/branch-2.6
Commit: 08db087b87d819814c2290120397775329c178ae
Parents: 536cfc3
Author: Nate Cole <nc...@hortonworks.com>
Authored: Tue Sep 12 16:04:16 2017 -0400
Committer: Nate Cole <nc...@hortonworks.com>
Committed: Wed Sep 13 09:58:27 2017 -0400
----------------------------------------------------------------------
.../ambari/server/stack/StackContext.java | 127 ++++++--
.../apache/ambari/server/stack/StackModule.java | 67 +++-
.../apache/ambari/server/state/StackInfo.java | 1 -
.../server/state/stack/LatestRepoCallable.java | 310 -------------------
.../server/state/stack/RepoUrlInfoCallable.java | 217 +++++++++++++
.../server/state/stack/RepoVdfCallable.java | 161 ++++++++++
.../ComponentVersionCheckActionTest.java | 14 -
7 files changed, 532 insertions(+), 365 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackContext.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/stack/StackContext.java b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackContext.java
index da7f021..e201e57 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/stack/StackContext.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackContext.java
@@ -18,20 +18,31 @@
package org.apache.ambari.server.stack;
-import org.apache.ambari.server.api.services.AmbariMetaInfo;
-import org.apache.ambari.server.metadata.ActionMetadata;
-import org.apache.ambari.server.orm.dao.MetainfoDAO;
-import org.apache.ambari.server.orm.entities.MetainfoEntity;
-import org.apache.ambari.server.state.stack.LatestRepoCallable;
-import org.apache.ambari.server.state.stack.OsFamily;
-
-import java.io.File;
+import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.metadata.ActionMetadata;
+import org.apache.ambari.server.orm.dao.MetainfoDAO;
+import org.apache.ambari.server.orm.entities.MetainfoEntity;
+import org.apache.ambari.server.state.stack.OsFamily;
+import org.apache.ambari.server.state.stack.RepoUrlInfoCallable;
+import org.apache.ambari.server.state.stack.RepoUrlInfoCallable.RepoUrlInfoResult;
+import org.apache.ambari.server.state.stack.RepoVdfCallable;
+import org.apache.commons.collections.MapUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* Provides external functionality to the Stack framework.
@@ -48,20 +59,18 @@ public class StackContext {
private ActionMetadata actionMetaData;
/**
- * Operating System families
- */
- private OsFamily osFamily;
-
- /**
* Executor used to get latest repo url's
*/
- private LatestRepoQueryExecutor repoUpdateExecutor = new LatestRepoQueryExecutor();
+ private LatestRepoQueryExecutor repoUpdateExecutor;
/**
* Repository XML base url property name
*/
private static final String REPOSITORY_XML_PROPERTY_BASEURL = "baseurl";
+ private final static Logger LOG = LoggerFactory.getLogger(StackContext.class);
+ private static final int THREAD_COUNT = 5;
+
/**
* Constructor.
@@ -73,7 +82,7 @@ public class StackContext {
public StackContext(MetainfoDAO metaInfoDAO, ActionMetadata actionMetaData, OsFamily osFamily) {
this.metaInfoDAO = metaInfoDAO;
this.actionMetaData = actionMetaData;
- this.osFamily = osFamily;
+ repoUpdateExecutor = new LatestRepoQueryExecutor(osFamily);
}
/**
@@ -109,11 +118,11 @@ public class StackContext {
* @param url external repo information URL
* @param stack stack module
*/
- public void registerRepoUpdateTask(String url, StackModule stack) {
- repoUpdateExecutor.addTask(new LatestRepoCallable(url,
- new File(stack.getStackDirectory().getRepoDir()), stack.getModuleInfo(), osFamily));
+ public void registerRepoUpdateTask(URI uri, StackModule stack) {
+ repoUpdateExecutor.addTask(uri, stack);
}
+
/**
* Execute the registered repo update tasks.
*/
@@ -139,16 +148,16 @@ public class StackContext {
/**
* Registered tasks
*/
- private Collection<LatestRepoCallable> tasks = new ArrayList<LatestRepoCallable>();
+ private Map<URI, RepoUrlInfoCallable> tasks = new HashMap<>();
/**
* Task futures
*/
- Collection<Future<Void>> futures = new ArrayList<Future<Void>>();
+ Collection<Future<?>> futures = new ArrayList<>();
/**
* Underlying executor
*/
- private ExecutorService executor = Executors.newSingleThreadExecutor(new ThreadFactory() {
+ private ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "Stack Version Loading Thread");
@@ -156,32 +165,92 @@ public class StackContext {
});
+ private OsFamily m_family;
+
+ private LatestRepoQueryExecutor(OsFamily family) {
+ m_family = family;
+ }
+
/**
- * Add a task.
- *
- * @param task task to be added
+ * @param uri
+ * uri to load
+ * @param stackModule
+ * the stack module
*/
- public void addTask(LatestRepoCallable task) {
- tasks.add(task);
+ public void addTask(URI uri, StackModule stackModule) {
+ RepoUrlInfoCallable callable = null;
+ if (tasks.containsKey(uri)) {
+ callable = tasks.get(uri);
+ } else {
+ callable = new RepoUrlInfoCallable(uri);
+ tasks.put(uri, callable);
+ }
+
+ callable.addStack(stackModule);
}
+
/**
* Execute all tasks.
*/
public void execute() {
- for (LatestRepoCallable task : tasks) {
- futures.add(executor.submit(task));
+
+ long l = System.nanoTime();
+ List<Future<Map<StackModule, RepoUrlInfoResult>>> results = new ArrayList<>();
+
+ // !!! first, load the *_urlinfo.json files and block for completion
+ try {
+ results = executor.invokeAll(tasks.values(), 2, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ LOG.warn("Could not load urlinfo as the executor was interrupted", e);
+ return;
+ } finally {
+ LOG.info("Loaded urlinfo in " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - l) + "ms");
}
+
+ List<Map<StackModule, RepoUrlInfoResult>> urlInfoResults = new ArrayList<>();
+ // !!! now load all the VDF _by version_ in a new thread.
+ for (Future<Map<StackModule, RepoUrlInfoResult>> future : results) {
+ try {
+ urlInfoResults.add(future.get());
+ } catch (Exception e) {
+ LOG.error("Could not load repo results", e.getCause());
+ }
+ }
+
+ for (Map<StackModule, RepoUrlInfoResult> urlInfoResult : urlInfoResults) {
+ for (Entry<StackModule, RepoUrlInfoResult> entry : urlInfoResult.entrySet()) {
+ StackModule stackModule = entry.getKey();
+ RepoUrlInfoResult result = entry.getValue();
+
+ if (null != result) {
+ if (MapUtils.isNotEmpty(result.getManifest())) {
+ for (Entry<String, Map<String, URI>> manifestEntry : result.getManifest().entrySet()) {
+ futures.add(executor.submit(new RepoVdfCallable(stackModule, manifestEntry.getKey(),
+ manifestEntry.getValue(), m_family)));
+ }
+ }
+
+ if (MapUtils.isNotEmpty(result.getLatestVdf())) {
+ futures.add(executor.submit(
+ new RepoVdfCallable(stackModule, result.getLatestVdf(), m_family)));
+ }
+ }
+ }
+ }
+
executor.shutdown();
}
+
+
/**
* Determine whether all tasks have completed.
*
* @return true if all tasks have completed; false otherwise
*/
public boolean hasCompleted() {
- for (Future<Void> f : futures) {
+ for (Future<?> f : futures) {
if (! f.isDone()) {
return false;
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackModule.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/stack/StackModule.java b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackModule.java
index cbbe92e..4f498d4 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/stack/StackModule.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/stack/StackModule.java
@@ -19,6 +19,8 @@
package org.apache.ambari.server.stack;
import java.io.File;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -101,17 +103,17 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
/**
* Map of child configuration modules keyed by configuration type
*/
- private Map<String, ConfigurationModule> configurationModules = new HashMap<String, ConfigurationModule>();
+ private Map<String, ConfigurationModule> configurationModules = new HashMap<>();
/**
* Map of child service modules keyed by service name
*/
- private Map<String, ServiceModule> serviceModules = new HashMap<String, ServiceModule>();
+ private Map<String, ServiceModule> serviceModules = new HashMap<>();
/**
* Map of linked extension modules keyed by extension name + version
*/
- private Map<String, ExtensionModule> extensionModules = new HashMap<String, ExtensionModule>();
+ private Map<String, ExtensionModule> extensionModules = new HashMap<>();
/**
* Corresponding StackInfo instance
@@ -321,7 +323,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
Collection<ServiceModule> mergedModules = mergeChildModules(
allStacks, commonServices, extensions, serviceModules, parentStack.serviceModules);
- List<String> removedServices = new ArrayList<String>();
+ List<String> removedServices = new ArrayList<>();
for (ServiceModule module : mergedModules) {
if (module.isDeleted()){
@@ -617,7 +619,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
* @param serviceDirectory the child service directory
*/
private void populateService(ServiceDirectory serviceDirectory) {
- Collection<ServiceModule> serviceModules = new ArrayList<ServiceModule>();
+ Collection<ServiceModule> serviceModules = new ArrayList<>();
// unfortunately, we allow multiple services to be specified in the same metainfo.xml,
// so we can't move the unmarshal logic into ServiceModule
ServiceMetainfoXml metaInfoXml = serviceDirectory.getMetaInfoFile();
@@ -769,7 +771,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
// Stack-definition has 'depends-on' relationship specified.
// We have a map to construct the 'depended-by' relationship.
Map<PropertyDependencyInfo, Set<PropertyDependencyInfo>> dependedByMap =
- new HashMap<PropertyDependencyInfo, Set<PropertyDependencyInfo>>();
+ new HashMap<>();
// Go through all service-configs and gather the reversed 'depended-by'
// relationship into map. Since we do not have the reverse {@link PropertyInfo},
@@ -785,7 +787,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
dependedByMap.get(pdi).add(propertyDependency);
} else {
Set<PropertyDependencyInfo> newDependenciesSet =
- new HashSet<PropertyDependencyInfo>();
+ new HashSet<>();
newDependenciesSet.add(propertyDependency);
dependedByMap.put(pdi, newDependenciesSet);
}
@@ -983,7 +985,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
if (skippedGroups.containsKey(group.addAfterGroup)) {
tmp = skippedGroups.get(group.addAfterGroup);
} else {
- tmp = new ArrayList<Grouping>();
+ tmp = new ArrayList<>();
skippedGroups.put(group.addAfterGroup, tmp);
}
tmp.add(group);
@@ -1140,10 +1142,53 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
stackInfo.getRepositories().addAll(serviceRepos);
if (null != rxml && null != rxml.getLatestURI() && stackRepos.size() > 0) {
- stackContext.registerRepoUpdateTask(rxml.getLatestURI(), this);
+ registerRepoUpdateTask(rxml);
}
}
+ private void registerRepoUpdateTask(RepositoryXml rxml) {
+ String latest = rxml.getLatestURI();
+ if (StringUtils.isBlank(latest)) {
+ return;
+ }
+
+ URI uri = getURI(this, latest);
+
+ if (null == uri) {
+ LOG.warn("Could not determine how to load stack {}-{} latest definition for {}",
+ stackInfo.getName(), stackInfo.getVersion(), latest);
+ return;
+ }
+
+ stackContext.registerRepoUpdateTask(uri, this);
+ }
+
+ /**
+ * @param module
+ * the stack module
+ * @param uriString
+ * the uri string
+ * @return a repo URI, even if it is relative-file based
+ */
+ public static URI getURI(StackModule module, String uriString) {
+
+ URI uri = null;
+ if (uriString.startsWith("http")) {
+ try {
+ uri = new URI(uriString);
+ } catch (URISyntaxException e) {
+ // should be logged later
+ }
+ } else if ('.' == uriString.charAt(0)) {
+ uri = new File(module.getStackDirectory().getRepoDir(), uriString).toURI();
+ } else {
+ uri = new File(uriString).toURI();
+ }
+
+ return uri;
+ }
+
+
/**
* Gets the service repos with duplicates filtered out. A service repo is considered duplicate if:
* <ul>
@@ -1230,7 +1275,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
if (null != serviceRepoXml) {
repos.addAll(serviceRepoXml.getRepositories());
if (null != serviceRepoXml.getLatestURI()) {
- stackContext.registerRepoUpdateTask(serviceRepoXml.getLatestURI(), this);
+ registerRepoUpdateTask(serviceRepoXml);
}
}
}
@@ -1299,7 +1344,7 @@ public class StackModule extends BaseModule<StackModule, StackInfo> implements V
this.valid = valid;
}
- private Set<String> errorSet = new HashSet<String>();
+ private Set<String> errorSet = new HashSet<>();
@Override
public void addError(String error) {
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/state/StackInfo.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/StackInfo.java b/ambari-server/src/main/java/org/apache/ambari/server/state/StackInfo.java
index f1412f5..b7f64f9 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/StackInfo.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/StackInfo.java
@@ -33,7 +33,6 @@ import org.apache.ambari.server.controller.StackVersionResponse;
import org.apache.ambari.server.stack.Validable;
import org.apache.ambari.server.state.repository.VersionDefinitionXml;
import org.apache.ambari.server.state.stack.ConfigUpgradePack;
-import org.apache.ambari.server.state.stack.LatestRepoCallable;
import org.apache.ambari.server.state.stack.RepositoryXml;
import org.apache.ambari.server.state.stack.StackRoleCommandOrder;
import org.apache.ambari.server.state.stack.UpgradePack;
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/LatestRepoCallable.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/LatestRepoCallable.java b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/LatestRepoCallable.java
deleted file mode 100644
index c43ce7c..0000000
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/LatestRepoCallable.java
+++ /dev/null
@@ -1,310 +0,0 @@
-/**
- * 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.ambari.server.state.stack;
-
-import java.io.File;
-import java.io.FileReader;
-import java.io.InputStreamReader;
-import java.lang.reflect.Type;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.Callable;
-
-import org.apache.ambari.server.controller.internal.URLStreamProvider;
-import org.apache.ambari.server.state.RepositoryInfo;
-import org.apache.ambari.server.state.StackId;
-import org.apache.ambari.server.state.StackInfo;
-import org.apache.ambari.server.state.repository.VersionDefinitionXml;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.gson.Gson;
-import com.google.gson.reflect.TypeToken;
-
-/**
- * Encapsulates the work to resolve the latest repo information for a stack.
- * This class must be used AFTER the stack has created its owned repositories.
- */
-public class LatestRepoCallable implements Callable<Void> {
- private static final int LOOKUP_CONNECTION_TIMEOUT = 2000;
- private static final int LOOKUP_READ_TIMEOUT = 3000;
-
- private final static Logger LOG = LoggerFactory.getLogger(LatestRepoCallable.class);
-
- private String sourceUri = null;
- private File stackRepoFolder = null;
- private StackInfo stack = null;
- private OsFamily os_family;
-
- public LatestRepoCallable(String latestSourceUri, File stackRepoFolder, StackInfo stack, OsFamily os_family) {
- this.sourceUri = latestSourceUri;
- this.stackRepoFolder = stackRepoFolder;
- this.stack = stack;
- this.os_family = os_family;
- }
-
- @Override
- public Void call() throws Exception {
-
- Type type = new TypeToken<Map<String, Map<String, Object>>>(){}.getType();
- Gson gson = new Gson();
-
- Map<String, Map<String, Object>> latestUrlMap = null;
-
- Long time = System.currentTimeMillis();
- try {
- if (sourceUri.startsWith("http")) {
-
- URLStreamProvider streamProvider = new URLStreamProvider(
- LOOKUP_CONNECTION_TIMEOUT, LOOKUP_READ_TIMEOUT,
- null, null, null);
-
- LOG.info("Loading latest URL info for stack {}-{} from {}", stack.getName(),
- stack.getVersion(), sourceUri);
-
- latestUrlMap = gson.fromJson(new InputStreamReader(
- streamProvider.readFrom(sourceUri)), type);
- } else {
- File jsonFile = null;
- if (sourceUri.charAt(0) == '.') {
- jsonFile = new File(stackRepoFolder, sourceUri);
- } else {
- jsonFile = new File(sourceUri);
- }
-
- if (jsonFile.exists()) {
- LOG.info("Loading latest URL info for stack {}-{} from {}", stack.getName(),
- stack.getVersion(), jsonFile);
- latestUrlMap = gson.fromJson(new FileReader(jsonFile), type);
- }
- }
- } catch (Exception e) {
- LOG.info("Could not load the URI for stack {}-{} from {}, ({}). Using default repository values",
- stack.getName(), stack.getVersion(), sourceUri, e.getMessage());
- throw e;
- } finally {
- LOG.info("Loaded uri {} in {}ms", sourceUri, System.currentTimeMillis() - time);
- }
-
- // !!! process latest overrides
- if (null != latestUrlMap) {
- for (RepositoryInfo ri : stack.getRepositories()) {
- if (latestUrlMap.containsKey(ri.getRepoId())) {
-
- Map<String, Object> valueMap = latestUrlMap.get(ri.getRepoId());
-
- if (valueMap.containsKey("latest")) {
-
- @SuppressWarnings("unchecked")
- Map<String, String> osMap = (Map<String, String>) valueMap.get("latest");
-
- String baseUrl = resolveOsUrl(ri.getOsType(), osMap);
- if (null != baseUrl) {
- // !!! in the case where <name>.repo is defined with the base url, strip that off.
- // Agents do the reverse action (take the base url, and append <name>.repo)
-
- String repo_file_format;
-
- if(os_family.isUbuntuFamily(ri.getOsType())) {
- repo_file_format = "list";
- } else {
- repo_file_format = "repo";
- }
-
- String repoFileName = stack.getName().toLowerCase() + "." + repo_file_format;
- int idx = baseUrl.toLowerCase().indexOf(repoFileName);
-
- if (-1 != idx && baseUrl.toLowerCase().endsWith(repoFileName)) {
- baseUrl = baseUrl.substring(0, idx);
- }
-
- if ('/' == baseUrl.charAt(baseUrl.length()-1)) {
- baseUrl = baseUrl.substring(0, baseUrl.length()-1);
- }
-
- ri.setLatestBaseUrl(baseUrl);
- if (ri.getBaseUrl() != null && !ri.isBaseUrlFromSaved()) {
- // Override baseUrl with the latestBaseUrl.
- ri.setBaseUrl(baseUrl);
- }
- }
- }
- }
- }
- }
-
- StackId stackId = new StackId(stack);
- if (!latestUrlMap.containsKey(stackId.toString())) {
- return null;
- }
-
- Map<String, Object> map = latestUrlMap.get(stackId.toString());
- if (null == map) {
- return null;
- }
-
- // !!! use this to prevent double loading of VDF.
- Map<URI, VersionDefinitionXml> parsedMap = new HashMap<>();
-
- if (map.containsKey("manifests")) {
-
- @SuppressWarnings("unchecked")
- Map<String, Object> versionMap = (Map<String, Object>) map.get("manifests");
-
- // EACH VDF is for ONLY ONE repository. We must provide a merged view.
- // there is no good way around this, so we have to make some concessions
-
- // !!! each key is a version number, and the value is a map containing
- // os_family -> VDF link
-
- for (Entry<String, Object> entry : versionMap.entrySet()) {
- String version = entry.getKey();
-
- @SuppressWarnings("unchecked")
- Map<String, String> osMap = (Map<String, String>) entry.getValue();
-
- VersionDefinitionXml xml = mergeDefinitions(stackId, version, osMap, parsedMap);
-
- if (null != xml) {
- stack.addVersionDefinition(version, xml);
- }
- }
- }
-
- if (map.containsKey("latest-vdf")) {
- // !!! this is an os_family -> VDF link. It's identical to the version entry in the 'manifests'
- // structure. This is the source of truth for the default, unspecified version when
- // doing a bluerpint install
- @SuppressWarnings("unchecked")
- Map<String, String> osMap = (Map<String, String>) map.get("latest-vdf");
-
- VersionDefinitionXml xml = mergeDefinitions(stackId, null, osMap, parsedMap);
- xml.setStackDefault(true);
- stack.setLatestVersionDefinition(xml);
- }
-
- return null;
- }
-
- /**
- * Merges definitions loaded from the common file
- * @param stackId the stack id
- * @param version the version string
- * @param osMap the map containing all the VDF for an OS
- * @return the merged version definition
- * @throws Exception
- */
- private VersionDefinitionXml mergeDefinitions(StackId stackId, String version,
- Map<String, String> osMap, Map<URI, VersionDefinitionXml> parsedMap) throws Exception {
-
- Set<String> oses = new HashSet<>();
- for (RepositoryInfo ri : stack.getRepositories()) {
- if (null != os_family.find(ri.getOsType())) {
- oses.add(os_family.find(ri.getOsType()));
- }
- }
-
- VersionDefinitionXml.Merger merger = new VersionDefinitionXml.Merger();
-
- for (Entry<String, String> versionEntry : osMap.entrySet()) {
-
- String osFamily = os_family.find(versionEntry.getKey());
-
- // !!! check for aliases. Moving this to OsFamily could result in incorrect behavior
- if (null == osFamily) {
- String alias = os_family.getAliases().get(versionEntry.getKey());
- if (null != alias) {
- osFamily = os_family.find(alias);
- }
- }
-
- // !!! if the family is not known OR not part of the stack, skip
- if (null == osFamily || !oses.contains(osFamily)) {
- LOG.info("Stack {} cannot resolve OS {} to the supported ones: {}. Family: {}",
- stackId, versionEntry.getKey(), StringUtils.join(oses, ','), osFamily);
- continue;
- }
-
- String uriString = versionEntry.getValue();
-
- if ('.' == uriString.charAt(0)) {
- uriString = new File(stackRepoFolder, uriString).toURI().toString();
- }
-
- try {
- URI uri = new URI(uriString);
-
- VersionDefinitionXml xml = parsedMap.containsKey(uri) ? parsedMap.get(uri) :
- timedVDFLoad(uri);
-
- version = (null == version) ? xml.release.version : version;
- merger.add(version, xml);
-
- if (!parsedMap.containsKey(uri)) {
- parsedMap.put(uri, xml);
- }
-
- } catch (Exception e) {
- LOG.warn("Could not load version definition for {} identified by {}. {}",
- stackId, uriString, e.getMessage(), e);
- }
- }
-
- return merger.merge();
- }
-
- /**
- * Resolves a base url given that certain OS types can be used interchangeably.
- * @param os the target os to find
- * @param osMap the map of os-to-baseurl
- * @return the url for an os.
- */
- private String resolveOsUrl(String os, Map<String, String> osMap) {
-
- // !!! look for the OS directly
- if (osMap.containsKey(os))
- return osMap.get(os);
-
- // !!! os not found, find and return the first compatible one
- Set<String> possibleTypes = os_family.findTypes(os);
-
- for (String type : possibleTypes) {
- if (osMap.containsKey(type))
- return osMap.get(type);
- }
-
- return null;
- }
-
- private VersionDefinitionXml timedVDFLoad(URI uri) throws Exception {
- long time = System.currentTimeMillis();
-
- try {
- return VersionDefinitionXml.load(uri.toURL());
- } finally {
- LOG.info("Loaded VDF {} in {}ms", uri, System.currentTimeMillis() - time);
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoUrlInfoCallable.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoUrlInfoCallable.java b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoUrlInfoCallable.java
new file mode 100644
index 0000000..ec1071e
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoUrlInfoCallable.java
@@ -0,0 +1,217 @@
+/**
+ * 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.ambari.server.state.stack;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.InputStreamReader;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.ambari.server.controller.internal.URLStreamProvider;
+import org.apache.ambari.server.stack.StackModule;
+import org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.stack.RepoUrlInfoCallable.RepoUrlInfoResult;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+
+/**
+ * Encapsulates the work to resolve the latest repo information for a stack.
+ * This class must be used AFTER the stack has created its owned repositories.
+ */
+public class RepoUrlInfoCallable implements Callable<Map<StackModule, RepoUrlInfoResult>> {
+ private static final int LOOKUP_CONNECTION_TIMEOUT = 2000;
+ private static final int LOOKUP_READ_TIMEOUT = 3000;
+
+ private final static Logger LOG = LoggerFactory.getLogger(RepoUrlInfoCallable.class);
+
+ private URI m_uri = null;
+ private Set<StackModule> m_stacks = new HashSet<>();
+
+ public RepoUrlInfoCallable(URI uri) {
+ m_uri = uri;
+ }
+
+ public void addStack(StackModule stackModule) {
+ m_stacks.add(stackModule);
+ }
+
+ @Override
+ public Map<StackModule, RepoUrlInfoResult> call() throws Exception {
+
+ Type type = new TypeToken<Map<String, Map<String, Object>>>(){}.getType();
+ Gson gson = new Gson();
+
+ Map<String, Map<String, Object>> latestUrlMap = null;
+
+ Set<String> ids = new HashSet<>();
+ ids.addAll(Collections2.transform(m_stacks, new Function<StackModule, String>() {
+ @Override
+ public String apply(StackModule input) {
+ // TODO Auto-generated method stub
+ return new StackId(input.getModuleInfo()).toString();
+ }
+ }));
+
+ String stackIds = StringUtils.join(ids, ',');
+
+ Long time = System.nanoTime();
+
+ try {
+ if (m_uri.getScheme().startsWith("http")) {
+ URLStreamProvider streamProvider = new URLStreamProvider(
+ LOOKUP_CONNECTION_TIMEOUT, LOOKUP_READ_TIMEOUT,
+ null, null, null);
+
+ LOG.info("Loading latest URL info from {} for stacks {}", m_uri, stackIds);
+
+ latestUrlMap = gson.fromJson(new InputStreamReader(
+ streamProvider.readFrom(m_uri.toString())), type);
+ } else {
+ File jsonFile = new File(m_uri);
+
+ if (jsonFile.exists()) {
+ LOG.info("Loading latest URL info from file {} for stacks {}", m_uri, stackIds);
+ latestUrlMap = gson.fromJson(new FileReader(jsonFile), type);
+ }
+ }
+ } catch (Exception e) {
+ LOG.info("Could not load the URI from {}, stack defaults will be used", m_uri);
+ throw e;
+ } finally {
+ LOG.info("Loaded URI {} for stacks {} in {}ms", m_uri, stackIds,
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - time));
+ }
+
+ Map<StackModule, RepoUrlInfoResult> result = new HashMap<>();
+
+ if (null == latestUrlMap) {
+ LOG.error("Could not load latest data for URI {} and stacks {}", m_uri, stackIds);
+ return result;
+ }
+
+ for (StackModule stackModule : m_stacks) {
+ StackId stackId = new StackId(stackModule.getModuleInfo());
+
+ Map<String, Object> map = latestUrlMap.get(stackId.toString());
+
+ if (null == map) {
+ continue;
+ }
+
+ RepoUrlInfoResult res = new RepoUrlInfoResult();
+
+ if (map.containsKey("manifests")) {
+ // versionMap is laid out like [version -> [os -> VDF uri]]
+ @SuppressWarnings("unchecked")
+ Map<String, Map<String, String>> versionMap = (Map<String, Map<String, String>>) map.get("manifests");
+
+ for (Entry<String, Map<String, String>> versionEntry : versionMap.entrySet()) {
+ String version = versionEntry.getKey();
+ Map<String, URI> resolvedOsMap = resolveOsMap(stackModule, versionEntry.getValue());
+
+ res.addVersion(version, resolvedOsMap);
+ }
+ }
+
+
+ if (map.containsKey("latest-vdf")) {
+ @SuppressWarnings("unchecked")
+ Map<String, String> osMap = (Map<String, String>) map.get("latest-vdf");
+
+ Map<String, URI> resolvedOsMap = resolveOsMap(stackModule, osMap);
+
+ res.setLatest(resolvedOsMap);
+ }
+
+ result.put(stackModule, res);
+ }
+
+
+ return result;
+
+ }
+
+ private Map<String, URI> resolveOsMap(StackModule stackModule, Map<String, String> osMap) {
+
+ Map<String, URI> resolved = new HashMap<>();
+
+ for (Entry<String, String> osEntry : osMap.entrySet()) {
+
+ String uriString = osEntry.getValue();
+
+ URI uri = StackModule.getURI(stackModule, uriString);
+
+ if (null == uri) {
+ LOG.warn("Could not resolve URI {}", uriString);
+ } else {
+ resolved.put(osEntry.getKey(), uri);
+ }
+ }
+
+ return resolved;
+ }
+
+
+ /**
+ * Stores the results saved per StackModule
+ */
+ public static class RepoUrlInfoResult {
+
+ private Map<String, Map<String, URI>> versions = new HashMap<>();
+ private Map<String, URI> latestVdf = new HashMap<>();
+
+ private void addVersion(String version, Map<String, URI> vdfMap) {
+ versions.put(version, vdfMap);
+ }
+
+ private void setLatest(Map<String, URI> latestMap) {
+ latestVdf = latestMap;
+ }
+
+ /**
+ * Each version entry here should be loaded in it's entirety in a new thread
+ */
+ public Map<String, Map<String, URI>> getManifest() {
+
+ return versions;
+ }
+
+ /**
+ * @return the latest vdf map
+ */
+ public Map<String, URI> getLatestVdf() {
+
+ return latestVdf;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoVdfCallable.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoVdfCallable.java b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoVdfCallable.java
new file mode 100644
index 0000000..81657a7
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/RepoVdfCallable.java
@@ -0,0 +1,161 @@
+/**
+ * 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.ambari.server.state.stack;
+
+import java.net.URI;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.Callable;
+
+import org.apache.ambari.server.stack.StackModule;
+import org.apache.ambari.server.state.RepositoryInfo;
+import org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.StackInfo;
+import org.apache.ambari.server.state.repository.VersionDefinitionXml;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates the work to resolve the latest repo information for a stack.
+ * This class must be used AFTER the stack has created its owned repositories.
+ */
+public class RepoVdfCallable implements Callable<Void> {
+
+ private final static Logger LOG = LoggerFactory.getLogger(RepoVdfCallable.class);
+
+ // !!! these are required for this callable to work
+ private final StackInfo m_stack;
+ private final OsFamily m_family;
+ private final Map<String, URI> m_vdfMap;
+
+ // !!! determines if this is for manifests or latest-vdf
+ private String m_version;
+
+ public RepoVdfCallable(StackModule stackModule,
+ String version, Map<String, URI> vdfOsMap, OsFamily os_family) {
+ m_stack = stackModule.getModuleInfo();
+ m_family = os_family;
+ m_version = version;
+ m_vdfMap = vdfOsMap;
+ }
+
+ public RepoVdfCallable(StackModule stackModule,
+ Map<String, URI> vdfOsMap, OsFamily os_family) {
+ m_stack = stackModule.getModuleInfo();
+ m_family = os_family;
+ m_version = null;
+ m_vdfMap = vdfOsMap;
+ }
+
+ @Override
+ public Void call() throws Exception {
+ if (MapUtils.isEmpty(m_vdfMap)) {
+ return null;
+ }
+
+ boolean forLatest = (null == m_version);
+
+ StackId stackId = new StackId(m_stack);
+
+ VersionDefinitionXml xml = mergeDefinitions(stackId, m_version, m_vdfMap);
+
+ if (null == xml) {
+ return null;
+ }
+
+ if (forLatest) {
+ xml.setStackDefault(true);
+ m_stack.setLatestVersionDefinition(xml);
+ } else {
+ m_stack.addVersionDefinition(m_version, xml);
+ }
+
+ return null;
+ }
+
+ /**
+ * Merges definitions loaded from the common file
+ * @param stackId the stack id
+ * @param version the version string
+ * @param osMap the map containing all the VDF for an OS
+ * @return the merged version definition
+ * @throws Exception
+ */
+ private VersionDefinitionXml mergeDefinitions(StackId stackId, String version,
+ Map<String, URI> osMap) throws Exception {
+
+ Set<String> oses = new HashSet<>();
+ for (RepositoryInfo ri : m_stack.getRepositories()) {
+ if (null != m_family.find(ri.getOsType())) {
+ oses.add(m_family.find(ri.getOsType()));
+ }
+ }
+
+ VersionDefinitionXml.Merger merger = new VersionDefinitionXml.Merger();
+
+ for (Entry<String, URI> versionEntry : osMap.entrySet()) {
+
+ String osFamily = m_family.find(versionEntry.getKey());
+ URI uri = versionEntry.getValue();
+
+ // !!! check for aliases. Moving this to OsFamily could result in incorrect behavior
+ if (null == osFamily) {
+ String alias = m_family.getAliases().get(versionEntry.getKey());
+ if (null != alias) {
+ osFamily = m_family.find(alias);
+ }
+ }
+
+ // !!! if the family is not known OR not part of the stack, skip
+ if (null == osFamily || !oses.contains(osFamily)) {
+ LOG.info("Stack {} cannot resolve OS {} to the supported ones: {}. Family: {}",
+ stackId, versionEntry.getKey(), StringUtils.join(oses, ','), osFamily);
+ continue;
+ }
+
+ try {
+ VersionDefinitionXml xml = timedVDFLoad(uri);
+
+ version = (null == version) ? xml.release.version : version;
+ merger.add(version, xml);
+
+ } catch (Exception e) {
+ LOG.warn("Could not load version definition for {} identified by {}. {}",
+ stackId, uri.toString(), e.getMessage(), e);
+ }
+ }
+
+ return merger.merge();
+ }
+
+ private VersionDefinitionXml timedVDFLoad(URI uri) throws Exception {
+ long time = System.currentTimeMillis();
+
+ try {
+ return VersionDefinitionXml.load(uri.toURL());
+ } finally {
+ LOG.debug("Loaded VDF {} in {}ms", uri, System.currentTimeMillis() - time);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/08db087b/ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/ComponentVersionCheckActionTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/ComponentVersionCheckActionTest.java b/ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/ComponentVersionCheckActionTest.java
index 2e216e6..fd2b528 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/ComponentVersionCheckActionTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/ComponentVersionCheckActionTest.java
@@ -36,7 +36,6 @@ import org.apache.ambari.server.actionmanager.HostRoleCommandFactory;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.agent.CommandReport;
import org.apache.ambari.server.agent.ExecutionCommand;
-import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.OrmTestHelper;
@@ -56,7 +55,6 @@ import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigFactory;
import org.apache.ambari.server.state.Host;
-import org.apache.ambari.server.state.RepositoryInfo;
import org.apache.ambari.server.state.RepositoryVersionState;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
@@ -308,12 +306,6 @@ public class ComponentVersionCheckActionTest {
makeUpgradeCluster(sourceStack, sourceRepo, targetStack, targetRepo);
- // Verify the repo before calling Finalize
- AmbariMetaInfo metaInfo = m_injector.getInstance(AmbariMetaInfo.class);
-
- RepositoryInfo repo = metaInfo.getRepository(sourceStack.getStackName(), sourceStack.getStackVersion(), "redhat6", sourceStack.getStackId());
- assertEquals(HDP_211_CENTOS6_REPO_URL, repo.getBaseUrl());
-
// Finalize the upgrade
Map<String, String> commandParams = new HashMap<>();
ExecutionCommand executionCommand = new ExecutionCommand();
@@ -470,12 +462,6 @@ public class ComponentVersionCheckActionTest {
sch = createNewServiceComponentHost(cluster, "ZOOKEEPER", "ZOOKEEPER_SERVER", "h1");
sch.setVersion(HDP_2_1_1_1);
- // Verify the repo before calling Finalize
- AmbariMetaInfo metaInfo = m_injector.getInstance(AmbariMetaInfo.class);
-
- RepositoryInfo repo = metaInfo.getRepository(sourceStack.getStackName(), sourceStack.getStackVersion(), "redhat6", sourceStack.getStackId());
- assertEquals(HDP_211_CENTOS6_REPO_URL, repo.getBaseUrl());
-
// Finalize the upgrade
Map<String, String> commandParams = new HashMap<>();
ExecutionCommand executionCommand = new ExecutionCommand();