ServicesTezDistributedCacheCheck extends AbstractCheckDescriptor { @Override public void perform(PrerequisiteCheck prerequisiteCheck, PrereqCheckRequest request) throws AmbariException { String dfsProtocolsRegex = DFS_PROTOCOLS_REGEX_DEFAULT; PrerequisiteCheckConfig prerequisiteCheckConfig = request.getPrerequisiteCheckConfig(); Map<String, String> checkProperties = null; if(prerequisiteCheckConfig != null) { checkProperties = prerequisiteCheckConfig.getCheckProperties(this.getClass().getName()); } if(checkProperties != null && checkProperties.containsKey(DFS_PROTOCOLS_REGEX_PROPERTY_NAME)) { dfsProtocolsRegex = checkProperties.get(DFS_PROTOCOLS_REGEX_PROPERTY_NAME); } final String clusterName = request.getClusterName(); final Cluster cluster = clustersProvider.get().getCluster(clusterName); final String tezConfigType = "tez-site"; final String coreSiteConfigType = "core-site"; final Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); final DesiredConfig tezDesiredConfig = desiredConfigs.get(tezConfigType); final Config tezConfig = cluster.getConfig(tezConfigType, tezDesiredConfig.getTag()); final DesiredConfig coreSiteDesiredConfig = desiredConfigs.get(coreSiteConfigType); final Config coreSiteConfig = cluster.getConfig(coreSiteConfigType, coreSiteDesiredConfig.getTag()); final String libUris = tezConfig.getProperties().get("tez.lib.uris"); final String useHadoopLibs = tezConfig.getProperties().get("tez.use.cluster.hadoop-libs"); final String defaultFS = coreSiteConfig.getProperties().get("fs.defaultFS"); List<String> errorMessages = new ArrayList<String>(); if (libUris == null || libUris.isEmpty()) { errorMessages.add(getFailReason(KEY_LIB_URI_MISSING, prerequisiteCheck, request)); } if (useHadoopLibs == null || useHadoopLibs.isEmpty()) { errorMessages.add(getFailReason(KEY_USE_HADOOP_LIBS, prerequisiteCheck, request)); } if (!errorMessages.isEmpty()) { prerequisiteCheck.getFailedOn().add("TEZ"); prerequisiteCheck.setStatus(PrereqCheckStatus.FAIL); prerequisiteCheck.setFailReason(StringUtils.join(errorMessages, " ")); return; } if (!libUris.matches(dfsProtocolsRegex) && (defaultFS == null || !defaultFS.matches(dfsProtocolsRegex))) { errorMessages.add(getFailReason(KEY_LIB_NOT_DFS, prerequisiteCheck, request)); } if (!libUris.contains("tar.gz")) { errorMessages.add(getFailReason(KEY_LIB_NOT_TARGZ, prerequisiteCheck, request)); } if (Boolean.parseBoolean(useHadoopLibs)) { errorMessages.add(getFailReason(KEY_USE_HADOOP_LIBS_FALSE, prerequisiteCheck, request)); } if (!errorMessages.isEmpty()) { prerequisiteCheck.getFailedOn().add("TEZ"); prerequisiteCheck.setStatus(PrereqCheckStatus.FAIL); prerequisiteCheck.setFailReason(StringUtils.join(errorMessages, " ")); } } ServicesTezDistributedCacheCheck(); @Override boolean isApplicable(PrereqCheckRequest request); @Override void perform(PrerequisiteCheck prerequisiteCheck, PrereqCheckRequest request);  }
@Test public void testPerform() throws Exception { final Cluster cluster = Mockito.mock(Cluster.class); Mockito.when(cluster.getClusterId()).thenReturn(1L); Mockito.when(clusters.getCluster("cluster")).thenReturn(cluster); final DesiredConfig desiredConfig = Mockito.mock(DesiredConfig.class); Mockito.when(desiredConfig.getTag()).thenReturn("tag"); Map<String, DesiredConfig> configMap = new HashMap<String, DesiredConfig>(); configMap.put("tez-site", desiredConfig); configMap.put("core-site", desiredConfig); Mockito.when(cluster.getDesiredConfigs()).thenReturn(configMap); final Config config = Mockito.mock(Config.class); Mockito.when(cluster.getConfig(Mockito.anyString(), Mockito.anyString())).thenReturn(config); final Map<String, String> properties = new HashMap<String, String>(); Mockito.when(config.getProperties()).thenReturn(properties); PrerequisiteCheck check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "hdfs: properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "dfs: properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "dfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.log"); properties.put("tez.use.cluster.hadoop-libs", "false"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "true"); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, new PrereqCheckRequest("cluster")); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); }
@Test public void testPerformWithCheckConfig() throws Exception { final Cluster cluster = Mockito.mock(Cluster.class); Mockito.when(cluster.getClusterId()).thenReturn(1L); Mockito.when(clusters.getCluster("cluster")).thenReturn(cluster); final DesiredConfig desiredConfig = Mockito.mock(DesiredConfig.class); Mockito.when(desiredConfig.getTag()).thenReturn("tag"); Map<String, DesiredConfig> configMap = new HashMap<String, DesiredConfig>(); configMap.put("tez-site", desiredConfig); configMap.put("core-site", desiredConfig); Mockito.when(cluster.getDesiredConfigs()).thenReturn(configMap); final Config config = Mockito.mock(Config.class); Mockito.when(cluster.getConfig(Mockito.anyString(), Mockito.anyString())).thenReturn(config); final Map<String, String> properties = new HashMap<String, String>(); Mockito.when(config.getProperties()).thenReturn(properties); Map<String, String> checkProperties = new HashMap<String, String>(); checkProperties.put("dfs-protocols-regex","^([^:]*dfs|wasb|ecs):.*"); PrerequisiteCheckConfig prerequisiteCheckConfig = Mockito.mock(PrerequisiteCheckConfig.class); Mockito.when(prerequisiteCheckConfig.getCheckProperties( servicesTezDistributedCacheCheck.getClass().getName())).thenReturn(checkProperties); PrereqCheckRequest request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); PrerequisiteCheck check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "hdfs: properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "dfs: properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "wasb: properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "ecs: properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "dfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "wasb: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "ecs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.PASS, check.getStatus()); properties.put("fs.defaultFS", "anything"); properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.log"); properties.put("tez.use.cluster.hadoop-libs", "false"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); properties.put("fs.defaultFS", "hdfs: properties.put("tez.lib.uris", "/some/path/to/archive.tar.gz"); properties.put("tez.use.cluster.hadoop-libs", "true"); request = new PrereqCheckRequest("cluster"); request.setPrerequisiteCheckConfig(prerequisiteCheckConfig); check = new PrerequisiteCheck(null, null); servicesTezDistributedCacheCheck.perform(check, request); Assert.assertEquals(PrereqCheckStatus.FAIL, check.getStatus()); }