| /** |
| * 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 |
| * <p/> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p/> |
| * 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.metrics.core.timeline.discovery; |
| |
| import java.io.IOException; |
| import java.net.MalformedURLException; |
| import java.net.URISyntaxException; |
| import java.sql.Connection; |
| import java.sql.SQLException; |
| import java.sql.Statement; |
| 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.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| import java.util.stream.Collectors; |
| |
| import org.apache.ambari.metrics.core.timeline.MetricsSystemInitializationException; |
| import org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration; |
| import org.apache.ambari.metrics.core.timeline.uuid.MetricUuidGenStrategy; |
| import org.apache.ambari.metrics.core.timeline.uuid.Murmur3HashUuidGenStrategy; |
| import org.apache.ambari.metrics.core.timeline.uuid.TimelineMetricUuid; |
| import org.apache.commons.collections.CollectionUtils; |
| import org.apache.commons.collections.MapUtils; |
| import org.apache.commons.lang.ArrayUtils; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.metrics2.sink.timeline.MetadataException; |
| import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; |
| import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata; |
| import org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor; |
| import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric; |
| |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_COMPRESSION_SCHEME; |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_ENCODING_SCHEME; |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TRANSIENT_METRIC_PATTERNS; |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.METRICS_METADATA_SYNC_INIT_DELAY; |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.METRICS_METADATA_SYNC_SCHEDULE_DELAY; |
| import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_METADATA_FILTERS; |
| import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_HOSTED_APPS_METADATA_TABLE_SQL; |
| import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_INSTANCE_HOST_TABLE_SQL; |
| import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_METADATA_TABLE_SQL; |
| import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_ENCODING; |
| import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_TABLE_COMPRESSION; |
| import static org.apache.hadoop.metrics2.sink.timeline.TimelineMetricUtils.getJavaMetricPatterns; |
| import static org.apache.hadoop.metrics2.sink.timeline.TimelineMetricUtils.getJavaRegexFromSqlRegex; |
| |
| public class TimelineMetricMetadataManager { |
| private static final Log LOG = LogFactory.getLog(TimelineMetricMetadataManager.class); |
| // Cache all metadata on retrieval |
| private final Map<TimelineMetricMetadataKey, TimelineMetricMetadata> METADATA_CACHE = new ConcurrentHashMap<>(); |
| private final Map<TimelineMetricUuid, TimelineMetricMetadataKey> uuidKeyMap = new ConcurrentHashMap<>(); |
| // Map to lookup apps on a host |
| private final Map<String, TimelineMetricHostMetadata> HOSTED_APPS_MAP = new ConcurrentHashMap<>(); |
| private final Map<TimelineMetricUuid, String> uuidHostMap = new ConcurrentHashMap<>(); |
| private final Map<String, Set<String>> INSTANCE_HOST_MAP = new ConcurrentHashMap<>(); |
| // Sync only when needed |
| AtomicBoolean SYNC_HOSTED_APPS_METADATA = new AtomicBoolean(false); |
| AtomicBoolean SYNC_HOSTED_INSTANCES_METADATA = new AtomicBoolean(false); |
| |
| private MetricUuidGenStrategy uuidGenStrategy = new Murmur3HashUuidGenStrategy(); |
| public static final int TIMELINE_METRIC_UUID_LENGTH = 16; |
| public static int HOSTNAME_UUID_LENGTH = 4; |
| |
| //Transient metric patterns. No UUID management and aggregation for such metrics. |
| private List<String> transientMetricPatterns = new ArrayList<>(); |
| |
| // Single thread to sync back new writes to the store |
| private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(); |
| |
| private PhoenixHBaseAccessor hBaseAccessor; |
| private Configuration metricsConf; |
| |
| TimelineMetricMetadataSync metricMetadataSync; |
| // Filter metrics names matching given patterns, from metadata |
| private final List<String> metricNameFilters = new ArrayList<>(); |
| |
| // Test friendly construction since mock instrumentation is difficult to get |
| // working with hadoop mini cluster |
| public TimelineMetricMetadataManager(Configuration metricsConf, PhoenixHBaseAccessor hBaseAccessor) { |
| this.metricsConf = metricsConf; |
| this.hBaseAccessor = hBaseAccessor; |
| String patternStrings = metricsConf.get(TIMELINE_METRIC_METADATA_FILTERS); |
| if (!StringUtils.isEmpty(patternStrings)) { |
| metricNameFilters.addAll(Arrays.asList(patternStrings.split(","))); |
| } |
| |
| uuidGenStrategy = getUuidStrategy(metricsConf); |
| |
| String transientMetricPatternsString = metricsConf.get(TRANSIENT_METRIC_PATTERNS, StringUtils.EMPTY); |
| if (StringUtils.isNotEmpty(transientMetricPatternsString)) { |
| LOG.info("Skipping UUID for patterns : " + transientMetricPatternsString); |
| transientMetricPatterns.addAll(getJavaMetricPatterns(transientMetricPatternsString)); |
| } |
| } |
| |
| public TimelineMetricMetadataManager(PhoenixHBaseAccessor hBaseAccessor) throws MalformedURLException, URISyntaxException { |
| this(TimelineMetricConfiguration.getInstance().getMetricsConf(), hBaseAccessor); |
| } |
| |
| /** |
| * Initialize Metadata from the store |
| */ |
| public void initializeMetadata() { |
| initializeMetadata(true); |
| } |
| |
| /** |
| * Initialize Metadata from the store |
| */ |
| public void initializeMetadata(boolean scheduleMetadateSync) { |
| |
| //Create metadata schema |
| Connection conn = null; |
| Statement stmt = null; |
| |
| String encoding = metricsConf.get(HBASE_ENCODING_SCHEME, DEFAULT_ENCODING); |
| String compression = metricsConf.get(HBASE_COMPRESSION_SCHEME, DEFAULT_TABLE_COMPRESSION); |
| |
| try { |
| LOG.info("Initializing metrics metadata schema..."); |
| conn = hBaseAccessor.getConnectionRetryingOnException(); |
| stmt = conn.createStatement(); |
| |
| // Metadata |
| String metadataSql = String.format(CREATE_METRICS_METADATA_TABLE_SQL, |
| encoding, compression); |
| stmt.executeUpdate(metadataSql); |
| |
| String hostedAppSql = String.format(CREATE_HOSTED_APPS_METADATA_TABLE_SQL, |
| encoding, compression); |
| stmt.executeUpdate(hostedAppSql); |
| |
| //Host Instances table |
| String hostedInstancesSql = String.format(CREATE_INSTANCE_HOST_TABLE_SQL, |
| encoding, compression); |
| stmt.executeUpdate(hostedInstancesSql); |
| } catch (SQLException | InterruptedException sql) { |
| LOG.error("Error creating Metrics Schema in HBase using Phoenix.", sql); |
| throw new MetricsSystemInitializationException( |
| "Error creating Metrics Metadata Schema in HBase using Phoenix.", sql); |
| } finally { |
| if (stmt != null) { |
| try { |
| stmt.close(); |
| } catch (SQLException e) { |
| // Ignore |
| } |
| } |
| if (conn != null) { |
| try { |
| conn.close(); |
| } catch (SQLException e) { |
| // Ignore |
| } |
| } |
| } |
| |
| metricMetadataSync = new TimelineMetricMetadataSync(this); |
| // Schedule the executor to sync to store |
| if (scheduleMetadateSync) { |
| executorService.scheduleWithFixedDelay(metricMetadataSync, |
| metricsConf.getInt(METRICS_METADATA_SYNC_INIT_DELAY, 120), // 2 minutes |
| metricsConf.getInt(METRICS_METADATA_SYNC_SCHEDULE_DELAY, 300), // 5 minutes |
| TimeUnit.SECONDS); |
| } |
| // Read from store and initialize map |
| try { |
| Map<TimelineMetricMetadataKey, TimelineMetricMetadata> metadata = getMetadataFromStore(); |
| |
| LOG.info("Retrieved " + metadata.size() + ", metadata objects from store."); |
| // Store in the cache |
| METADATA_CACHE.putAll(metadata); |
| |
| Map<String, TimelineMetricHostMetadata> hostedAppData = getHostedAppsFromStore(); |
| |
| LOG.info("Retrieved " + hostedAppData.size() + " host objects from store."); |
| HOSTED_APPS_MAP.putAll(hostedAppData); |
| |
| loadUuidMapsOnInit(); |
| |
| hBaseAccessor.setMetadataInstance(this); |
| } catch (SQLException e) { |
| LOG.warn("Exception loading metric metadata", e); |
| } |
| } |
| |
| public Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getMetadataCache() { |
| return METADATA_CACHE; |
| } |
| |
| public TimelineMetricMetadata getMetadataCacheValue(TimelineMetricMetadataKey key) { |
| return METADATA_CACHE.get(key); |
| } |
| |
| public Map<String, TimelineMetricHostMetadata> getHostedAppsCache() { |
| return HOSTED_APPS_MAP; |
| } |
| |
| public Map<String, Set<String>> getHostedInstanceCache() { |
| return INSTANCE_HOST_MAP; |
| } |
| |
| public boolean syncHostedAppsMetadata() { |
| return SYNC_HOSTED_APPS_METADATA.get(); |
| } |
| |
| public boolean syncHostedInstanceMetadata() { |
| return SYNC_HOSTED_INSTANCES_METADATA.get(); |
| } |
| |
| public void markSuccessOnSyncHostedAppsMetadata() { |
| SYNC_HOSTED_APPS_METADATA.set(false); |
| } |
| |
| public void markSuccessOnSyncHostedInstanceMetadata() { |
| SYNC_HOSTED_INSTANCES_METADATA.set(false); |
| } |
| |
| /** |
| * Test metric name for valid patterns and return true/false |
| */ |
| boolean skipMetadataCache(String metricName) { |
| for (String pattern : metricNameFilters) { |
| if (metricName.contains(pattern)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Update value in metadata cache |
| * @param metadata @TimelineMetricMetadata |
| */ |
| public void putIfModifiedTimelineMetricMetadata(TimelineMetricMetadata metadata) { |
| if (skipMetadataCache(metadata.getMetricName())) { |
| return; |
| } |
| |
| TimelineMetricMetadataKey key = new TimelineMetricMetadataKey( |
| metadata.getMetricName(), metadata.getAppId(), metadata.getInstanceId()); |
| |
| TimelineMetricMetadata metadataFromCache = METADATA_CACHE.get(key); |
| |
| if (metadataFromCache != null) { |
| try { |
| if (metadataFromCache.needsToBeSynced(metadata)) { |
| metadata.setIsPersisted(false); // Set the flag to ensure sync to store on next run |
| METADATA_CACHE.put(key, metadata); |
| } |
| } catch (MetadataException e) { |
| LOG.warn("Error inserting Metadata in cache.", e); |
| } |
| |
| } else { |
| METADATA_CACHE.put(key, metadata); |
| } |
| } |
| |
| /** |
| * Update value in hosted apps cache |
| * @param hostname Host name |
| * @param appId Application Id |
| */ |
| public void putIfModifiedHostedAppsMetadata(String hostname, String appId) { |
| TimelineMetricHostMetadata timelineMetricHostMetadata = HOSTED_APPS_MAP.get(hostname); |
| ConcurrentHashMap<String, String> apps = (timelineMetricHostMetadata != null) ? timelineMetricHostMetadata.getHostedApps() : null; |
| if (apps == null) { |
| apps = new ConcurrentHashMap<>(); |
| if (timelineMetricHostMetadata == null) { |
| TimelineMetricHostMetadata newHostMetadata = new TimelineMetricHostMetadata(apps); |
| HOSTED_APPS_MAP.put(hostname, newHostMetadata); |
| } else { |
| HOSTED_APPS_MAP.get(hostname).setHostedApps(apps); |
| } |
| } |
| |
| if (!apps.containsKey(appId)) { |
| apps.put(appId, appId); |
| SYNC_HOSTED_APPS_METADATA.set(true); |
| } |
| } |
| |
| public void putIfModifiedHostedInstanceMetadata(String instanceId, String hostname) { |
| if (StringUtils.isEmpty(instanceId)) { |
| return; |
| } |
| |
| Set<String> hosts = INSTANCE_HOST_MAP.get(instanceId); |
| if (hosts == null) { |
| hosts = new HashSet<>(); |
| INSTANCE_HOST_MAP.put(instanceId, hosts); |
| } |
| |
| if (!hosts.contains(hostname)) { |
| hosts.add(hostname); |
| SYNC_HOSTED_INSTANCES_METADATA.set(true); |
| } |
| } |
| |
| public void persistMetadata(Collection<TimelineMetricMetadata> metadata) throws SQLException { |
| hBaseAccessor.saveMetricMetadata(metadata); |
| } |
| |
| public void persistHostedAppsMetadata(Map<String, TimelineMetricHostMetadata> hostedApps) throws SQLException { |
| hBaseAccessor.saveHostAppsMetadata(hostedApps); |
| } |
| |
| public void persistHostedInstanceMetadata(Map<String, Set<String>> hostedInstancesMetadata) throws SQLException { |
| hBaseAccessor.saveInstanceHostsMetadata(hostedInstancesMetadata); |
| } |
| |
| public TimelineMetricMetadata createTimelineMetricMetadata(TimelineMetric timelineMetric, boolean isWhitelisted) { |
| TimelineMetricMetadata timelineMetricMetadata = new TimelineMetricMetadata( |
| timelineMetric.getMetricName(), |
| timelineMetric.getAppId(), |
| timelineMetric.getInstanceId(), |
| timelineMetric.getUnits(), |
| timelineMetric.getType(), |
| timelineMetric.getStartTime(), |
| supportAggregates(timelineMetric), |
| isWhitelisted |
| ); |
| |
| //Set UUID for metadata on the write path. Do not pass in hostname here since we only want Metric metadata, not host metadata. |
| if (!isTransientMetric(timelineMetric.getMetricName(), timelineMetric.getAppId())) { |
| byte[] uuid = getUuid(timelineMetric.getMetricName(), timelineMetric.getAppId(), timelineMetric.getInstanceId(), null, true); |
| timelineMetricMetadata.setUuid(uuid); |
| } |
| return timelineMetricMetadata; |
| } |
| |
| boolean isDistributedModeEnabled() { |
| String mode = metricsConf.get("timeline.metrics.service.operation.mode"); |
| return (mode != null) && mode.equals("distributed"); |
| } |
| |
| /** |
| * Fetch metrics metadata from store from V1 table (no UUID) |
| * @throws SQLException |
| */ |
| Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getMetadataFromStoreV1() throws SQLException { |
| return hBaseAccessor.getTimelineMetricMetadataV1(); |
| } |
| |
| /** |
| * Fetch metrics metadata from store |
| * @throws SQLException |
| */ |
| Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getMetadataFromStore() throws SQLException { |
| return hBaseAccessor.getTimelineMetricMetadata(); |
| } |
| |
| /** |
| * Fetch hosted apps from store |
| * @throws SQLException |
| */ |
| Map<String, TimelineMetricHostMetadata> getHostedAppsFromStore() throws SQLException { |
| return hBaseAccessor.getHostedAppsMetadata(); |
| } |
| |
| Map<String, Set<String>> getHostedInstancesFromStore() throws SQLException { |
| return hBaseAccessor.getInstanceHostsMetdata(); |
| } |
| |
| private boolean supportAggregates(TimelineMetric metric) { |
| return MapUtils.isEmpty(metric.getMetadata()) || |
| !(String.valueOf(true).equals(metric.getMetadata().get("skipAggregation"))); |
| } |
| |
| ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// |
| // UUID Management |
| ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// |
| |
| /** |
| * Load the UUID mappings from the UUID table on startup. |
| */ |
| private void loadUuidMapsOnInit() { |
| |
| for (TimelineMetricMetadataKey key : METADATA_CACHE.keySet()) { |
| TimelineMetricMetadata timelineMetricMetadata = METADATA_CACHE.get(key); |
| if (timelineMetricMetadata != null && timelineMetricMetadata.getUuid() != null) { |
| uuidKeyMap.put(new TimelineMetricUuid(timelineMetricMetadata.getUuid()), key); |
| } |
| } |
| |
| for (String host : HOSTED_APPS_MAP.keySet()) { |
| TimelineMetricHostMetadata timelineMetricHostMetadata = HOSTED_APPS_MAP.get(host); |
| if (timelineMetricHostMetadata != null && timelineMetricHostMetadata.getUuid() != null) { |
| uuidHostMap.put(new TimelineMetricUuid(timelineMetricHostMetadata.getUuid()), host); |
| } |
| } |
| } |
| |
| /** |
| * Returns the UUID gen strategy. |
| * @param configuration the config |
| * @return the UUID generator of type org.apache.ambari.metrics.core.timeline.uuid.MetricUuidGenStrategy |
| */ |
| private MetricUuidGenStrategy getUuidStrategy(Configuration configuration) { |
| return new Murmur3HashUuidGenStrategy(); |
| } |
| |
| /** |
| * Given the hostname, generates a byte array of length 'HOSTNAME_UUID_LENGTH' |
| * @param hostname the hostname |
| * @param createIfNotPresent Generate UUID if not present. |
| * @return uuid byte array of length 'HOSTNAME_UUID_LENGTH' |
| */ |
| private byte[] getUuidForHostname(String hostname, boolean createIfNotPresent) { |
| |
| TimelineMetricHostMetadata timelineMetricHostMetadata = HOSTED_APPS_MAP.get(hostname); |
| if (timelineMetricHostMetadata != null) { |
| byte[] uuid = timelineMetricHostMetadata.getUuid(); |
| if (uuid != null) { |
| return uuid; |
| } |
| } |
| |
| if (!createIfNotPresent) { |
| LOG.debug("UUID not found for " + hostname + ", createIfNotPresent is false"); |
| return null; |
| } |
| |
| byte[] uuid = uuidGenStrategy.computeUuid(hostname, HOSTNAME_UUID_LENGTH); |
| TimelineMetricUuid timelineMetricUuid = new TimelineMetricUuid(uuid); |
| if (uuidHostMap.containsKey(timelineMetricUuid) && !hostname.equals(uuidHostMap.get(timelineMetricUuid))) { |
| LOG.error("Duplicate key computed for " + hostname +", Collides with " + uuidHostMap.get(timelineMetricUuid)); |
| return null; |
| } |
| |
| timelineMetricHostMetadata = HOSTED_APPS_MAP.computeIfAbsent(hostname, k -> new TimelineMetricHostMetadata()); |
| if (timelineMetricHostMetadata.getUuid() == null) { |
| timelineMetricHostMetadata.setUuid(uuid); |
| } |
| if (!uuidHostMap.containsKey(timelineMetricUuid)) { |
| uuidHostMap.put(timelineMetricUuid, hostname); |
| } |
| |
| return uuid; |
| } |
| |
| /** |
| * Given a timelineClusterMetric instance, generates a UUID for Metric-App-Instance combination. |
| * @param timelineClusterMetric The timeline cluster metric for which the UUID needs to be generated. |
| * @param createIfNotPresent Generate UUID if not present. |
| * @return uuid byte array of length 'TIMELINE_METRIC_UUID_LENGTH' |
| */ |
| public byte[] getUuid(TimelineClusterMetric timelineClusterMetric, boolean createIfNotPresent) { |
| |
| TimelineMetricMetadataKey key = new TimelineMetricMetadataKey(timelineClusterMetric.getMetricName(), |
| timelineClusterMetric.getAppId(), timelineClusterMetric.getInstanceId()); |
| |
| TimelineMetricMetadata timelineMetricMetadata = METADATA_CACHE.get(key); |
| if (timelineMetricMetadata != null) { |
| byte[] uuid = timelineMetricMetadata.getUuid(); |
| if (uuid != null) { |
| return uuid; |
| } |
| } |
| |
| if (!createIfNotPresent) { |
| LOG.debug("UUID not found for " + key + ", createIfNotPresent is false"); |
| return null; |
| } |
| |
| byte[] uuidBytes = uuidGenStrategy.computeUuid(timelineClusterMetric, TIMELINE_METRIC_UUID_LENGTH); |
| |
| TimelineMetricUuid uuid = new TimelineMetricUuid(uuidBytes); |
| if (uuidKeyMap.containsKey(uuid) && !uuidKeyMap.get(uuid).equals(key)) { |
| TimelineMetricMetadataKey collidingKey = uuidKeyMap.get(uuid); |
| LOG.error("Duplicate key " + uuid + " computed for " + timelineClusterMetric + ", Collides with " + collidingKey); |
| return null; |
| } |
| |
| timelineMetricMetadata = METADATA_CACHE.get(key); |
| if (timelineMetricMetadata == null) { |
| timelineMetricMetadata = new TimelineMetricMetadata(); |
| timelineMetricMetadata.setMetricName(timelineClusterMetric.getMetricName()); |
| timelineMetricMetadata.setAppId(timelineClusterMetric.getAppId()); |
| timelineMetricMetadata.setInstanceId(timelineClusterMetric.getInstanceId()); |
| METADATA_CACHE.put(key, timelineMetricMetadata); |
| } |
| |
| if (timelineMetricMetadata.getUuid() == null) { |
| timelineMetricMetadata.setUuid(uuid.uuid); |
| } |
| timelineMetricMetadata.setIsPersisted(false); |
| |
| if (!uuidKeyMap.containsKey(uuid)) { |
| uuidKeyMap.put(uuid, key); |
| } |
| return uuid.uuid; |
| } |
| |
| /** |
| * Given a timelineMetric instance, generates a UUID for Metric-App-Instance-Host combination. |
| * @param timelineMetric The timeline metric for which the UUID needs to be generated. |
| * @param createIfNotPresent Generate UUID if not present. |
| * @return uuid byte array of length 'TIMELINE_METRIC_UUID_LENGTH' + 'HOSTNAME_UUID_LENGTH' |
| */ |
| public byte[] getUuid(TimelineMetric timelineMetric, boolean createIfNotPresent) { |
| |
| byte[] metricUuid = getUuid(new TimelineClusterMetric(timelineMetric.getMetricName(), timelineMetric.getAppId(), |
| timelineMetric.getInstanceId(), -1l), createIfNotPresent); |
| byte[] hostUuid = getUuidForHostname(timelineMetric.getHostName(), createIfNotPresent); |
| |
| if (hostUuid != null) { |
| putIfModifiedHostedAppsMetadata(timelineMetric.getHostName(), timelineMetric.getAppId()); |
| } |
| |
| if (metricUuid == null || hostUuid == null) { |
| return null; |
| } |
| return ArrayUtils.addAll(metricUuid, hostUuid); |
| } |
| |
| /** |
| * Given a metric name, appId, instanceId and hotname, generates a UUID for Metric-App-Instance-Host combination. |
| * @param createIfNotPresent Generate UUID if not present. |
| * @return uuid byte array of length 'TIMELINE_METRIC_UUID_LENGTH' + 'HOSTNAME_UUID_LENGTH' |
| */ |
| public byte[] getUuid(String metricName, String appId, String instanceId, String hostname, boolean createIfNotPresent) { |
| |
| byte[] metricUuid = getUuid(new TimelineClusterMetric(metricName, appId, instanceId, -1l), createIfNotPresent); |
| if (StringUtils.isNotEmpty(hostname)) { |
| byte[] hostUuid = getUuidForHostname(hostname, createIfNotPresent); |
| if (hostUuid == null || metricUuid == null) { |
| return null; |
| } |
| return ArrayUtils.addAll(metricUuid, hostUuid); |
| } |
| return metricUuid; |
| } |
| |
| public String getMetricNameFromUuid(byte[] uuid) { |
| |
| byte[] metricUuid = uuid; |
| if (uuid.length == TIMELINE_METRIC_UUID_LENGTH + HOSTNAME_UUID_LENGTH) { |
| metricUuid = ArrayUtils.subarray(uuid, 0, TIMELINE_METRIC_UUID_LENGTH); |
| } |
| |
| TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(metricUuid)); |
| return key != null ? key.getMetricName() : null; |
| } |
| |
| /** |
| * Given a UUID (from DB hopefully), return the timeline metric it is associated with. |
| * @param uuid 'TIMELINE_METRIC_UUID_LENGTH' + 'HOSTNAME_UUID_LENGTH' byte UUID. |
| * @return TimelineMetric object if present in the metadata. |
| */ |
| public TimelineMetric getMetricFromUuid(byte[] uuid) { |
| if (uuid == null) { |
| return null; |
| } |
| |
| if (uuid.length == TIMELINE_METRIC_UUID_LENGTH) { |
| TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(uuid)); |
| return key != null ? new TimelineMetric(key.metricName, null, key.appId, key.instanceId) : null; |
| } else { |
| byte[] metricUuid = ArrayUtils.subarray(uuid, 0, TIMELINE_METRIC_UUID_LENGTH); |
| TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(metricUuid)); |
| if (key == null) { |
| LOG.error("TimelineMetricMetadataKey is null for : " + Arrays.toString(uuid)); |
| return null; |
| } |
| TimelineMetric timelineMetric = new TimelineMetric(); |
| timelineMetric.setMetricName(key.metricName); |
| timelineMetric.setAppId(key.appId); |
| timelineMetric.setInstanceId(key.instanceId); |
| |
| byte[] hostUuid = ArrayUtils.subarray(uuid, TIMELINE_METRIC_UUID_LENGTH, HOSTNAME_UUID_LENGTH + TIMELINE_METRIC_UUID_LENGTH); |
| String hostname = uuidHostMap.get(new TimelineMetricUuid(hostUuid)); |
| if (hostname == null) { |
| return null; |
| } |
| timelineMetric.setHostName(hostname); |
| return timelineMetric; |
| } |
| } |
| |
| public List<byte[]> getUuidsForGetMetricQuery(Collection<String> metricNames, |
| List<String> hostnames, |
| String appId, |
| String instanceId) { |
| return getUuidsForGetMetricQuery(metricNames, hostnames, appId, instanceId, Collections.EMPTY_LIST); |
| } |
| /** |
| * Returns the set of UUIDs for a given GET request. If there are wildcards (%), resolves them based on UUID map. |
| * If metricName-App-Instance or hostname not present in Metadata, the combination will be skipped. |
| * @param metricNames |
| * @param hostnames |
| * @param appId |
| * @param instanceId |
| * @return Set of UUIds |
| */ |
| public List<byte[]> getUuidsForGetMetricQuery(Collection<String> metricNames, |
| List<String> hostnames, |
| String appId, |
| String instanceId, |
| List<String> transientMetricNames) { |
| |
| List<byte[]> uuids = new ArrayList<>(); |
| |
| boolean metricNameHasWildcard = false; |
| for (String metricName : metricNames) { |
| if (hasWildCard(metricName)) { |
| metricNameHasWildcard = true; |
| break; |
| } |
| } |
| |
| boolean hostNameHasWildcard = false; |
| if (CollectionUtils.isNotEmpty(hostnames)) { |
| for (String hostname : hostnames) { |
| if (hasWildCard(hostname)) { |
| hostNameHasWildcard = true; |
| break; |
| } |
| } |
| } |
| |
| if ( StringUtils.isNotEmpty(appId) && !(appId.equals("HOST") || appId.equals("FLUME_HANDLER"))) { //HACK.. Why?? |
| appId = appId.toLowerCase(); |
| } |
| |
| if (hasWildCard(instanceId) || hasWildCard(appId) || hostNameHasWildcard || metricNameHasWildcard) { |
| try { |
| List<TimelineMetricMetadata> metricMetadataFromStore = hBaseAccessor.scanMetricMetadataForWildCardRequest(metricNames, |
| appId, instanceId); |
| List<byte[]> hostUuidsFromStore = hBaseAccessor.scanHostMetadataForWildCardRequest(hostnames); |
| |
| for (TimelineMetricMetadata matchedEntry : metricMetadataFromStore) { |
| if (matchedEntry.getUuid() != null) { |
| if (CollectionUtils.isNotEmpty(hostnames)) { |
| for (byte[] hostUuidEntry : hostUuidsFromStore) { |
| uuids.add(ArrayUtils.addAll(matchedEntry.getUuid(), hostUuidEntry)); |
| } |
| } else { |
| uuids.add(matchedEntry.getUuid()); |
| } |
| } else if (isTransientMetric(matchedEntry.getMetricName(), matchedEntry.getAppId())) { |
| transientMetricNames.add(matchedEntry.getMetricName()); |
| } |
| } |
| return uuids; |
| } catch (SQLException e) { |
| LOG.error("Unable to query metadata table to check satisfying metric keys for wildcard request : " + e); |
| return uuids; |
| } |
| } else { |
| |
| if (CollectionUtils.isNotEmpty(hostnames)) { |
| if (CollectionUtils.isNotEmpty(metricNames)) { |
| //Skip getting UUID if it is a transient metric. |
| //An attempt to get it will also be OK as we don't add null UUIDs. |
| for (String metricName : metricNames) { |
| if (isTransientMetric(metricName, appId)) { |
| transientMetricNames.add(metricName); |
| continue; |
| } |
| TimelineMetric metric = new TimelineMetric(); |
| metric.setMetricName(metricName); |
| metric.setAppId(appId); |
| metric.setInstanceId(instanceId); |
| for (String hostname : hostnames) { |
| metric.setHostName(hostname); |
| byte[] uuid = getUuid(metric, false); |
| if (uuid != null) { |
| uuids.add(uuid); |
| } |
| } |
| } |
| } else { |
| for (String hostname : hostnames) { |
| byte[] uuid = getUuidForHostname(hostname, false); |
| if (uuid != null) { |
| uuids.add(uuid); |
| } |
| } |
| } |
| } else { |
| for (String metricName : metricNames) { |
| //Skip getting UUID if it is a transient metric. An attempt to get it will also be OK as we don't add null UUIDs. |
| if (isTransientMetric(metricName, appId)) { |
| transientMetricNames.add(metricName); |
| continue; |
| } |
| TimelineClusterMetric metric = new TimelineClusterMetric(metricName, appId, instanceId, -1l); |
| byte[] uuid = getUuid(metric, false); |
| if (uuid != null) { |
| uuids.add(uuid); |
| } |
| } |
| } |
| } |
| |
| return uuids; |
| } |
| |
| ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// |
| |
| public List<String> getNotLikeHostnames(List<String> hostnames) { |
| List<String> result = new ArrayList<>(); |
| Set<String> sanitizedHostNames = getSanitizedHostnames(hostnames); |
| for (String hostname: HOSTED_APPS_MAP.keySet()) { |
| if (!sanitizedHostNames.contains(hostname)) { |
| result.add(hostname); |
| } |
| } |
| return result; |
| } |
| |
| private Set<String> getSanitizedHostnames(List<String> hostnamedWithOrWithoutWildcard) { |
| |
| Set<String> sanitizedHostNames = new HashSet<>(); |
| if (CollectionUtils.isNotEmpty(hostnamedWithOrWithoutWildcard)) { |
| for (String hostname : hostnamedWithOrWithoutWildcard) { |
| if (hostname.contains("%")) { |
| String hostRegEx; |
| hostRegEx = hostname.replace("%", ".*"); |
| for (String host : HOSTED_APPS_MAP.keySet()) { |
| if (host.matches(hostRegEx)) { |
| sanitizedHostNames.add(host); |
| } |
| } |
| } else { |
| sanitizedHostNames.add(hostname); |
| } |
| } |
| } |
| return sanitizedHostNames; |
| } |
| |
| /** |
| * |
| * @param appId |
| * @param metricPattern |
| * @param includeBlacklistedMetrics |
| * @return |
| * @throws SQLException |
| * @throws IOException |
| */ |
| public Map<String, List<TimelineMetricMetadata>> getTimelineMetricMetadataByAppId(String appId, String metricPattern, |
| boolean includeBlacklistedMetrics) throws SQLException, IOException { |
| |
| Map<TimelineMetricMetadataKey, TimelineMetricMetadata> metadata = getMetadataCache(); |
| |
| boolean filterByAppId = StringUtils.isNotEmpty(appId); |
| boolean filterByMetricName = StringUtils.isNotEmpty(metricPattern); |
| Pattern metricFilterPattern = null; |
| if (filterByMetricName) { |
| metricFilterPattern = Pattern.compile(metricPattern); |
| } |
| |
| // Group Metadata by AppId |
| Map<String, List<TimelineMetricMetadata>> metadataByAppId = new HashMap<>(); |
| for (TimelineMetricMetadata metricMetadata : metadata.values()) { |
| |
| if (!includeBlacklistedMetrics && !metricMetadata.isWhitelisted()) { |
| continue; |
| } |
| |
| String currentAppId = metricMetadata.getAppId(); |
| if (filterByAppId && !currentAppId.equals(appId)) { |
| continue; |
| } |
| |
| if (filterByMetricName) { |
| Matcher m = metricFilterPattern.matcher(metricMetadata.getMetricName()); |
| if (!m.find()) { |
| continue; |
| } |
| } |
| |
| List<TimelineMetricMetadata> metadataList = metadataByAppId.get(currentAppId); |
| if (metadataList == null) { |
| metadataList = new ArrayList<>(); |
| metadataByAppId.put(currentAppId, metadataList); |
| } |
| |
| metadataList.add(metricMetadata); |
| } |
| |
| return metadataByAppId; |
| } |
| |
| /** |
| * Returns metadata summary |
| * @return |
| * @throws IOException |
| * @throws SQLException |
| */ |
| public Map<String, String> getMetadataSummary() throws IOException, SQLException { |
| Map<String, String> summary = new HashMap<>(); |
| summary.put("Number of Hosts", String.valueOf(HOSTED_APPS_MAP.size())); |
| Map<String, List<TimelineMetricMetadata>> metadataMap = getTimelineMetricMetadataByAppId(StringUtils.EMPTY, |
| StringUtils.EMPTY, |
| true); |
| |
| if (metadataMap != null) { |
| for (String appId : metadataMap.keySet()) { |
| summary.put(appId, String.valueOf(metadataMap.get(appId).size())); |
| } |
| } |
| return summary; |
| } |
| |
| |
| /** |
| * |
| * @param metricName |
| * @param appId |
| * @return |
| */ |
| public boolean isTransientMetric(String metricName, String appId) { |
| //Currently we use only metric name. In the future we may use appId as well. |
| |
| for (String pattern : transientMetricPatterns) { |
| if (metricName.matches(pattern)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Run TimelineMetricMetadataSync once |
| */ |
| public void forceMetricsMetadataSync() { |
| if (metricMetadataSync == null) { |
| metricMetadataSync = new TimelineMetricMetadataSync(this); |
| } |
| metricMetadataSync.run(); |
| } |
| |
| public void updateMetadataCacheUsingV1Tables() throws SQLException { |
| Map<TimelineMetricMetadataKey, TimelineMetricMetadata> metadataV1Map = getMetadataFromStoreV1(); |
| for (TimelineMetricMetadataKey key: METADATA_CACHE.keySet()) { |
| TimelineMetricMetadata cacheValue = METADATA_CACHE.get(key); |
| TimelineMetricMetadata oldValue = metadataV1Map.get(key); |
| |
| if (oldValue != null) { |
| if (!cacheValue.isPersisted()) { |
| LOG.info(String.format("Updating properties for %s", key)); |
| cacheValue.setSeriesStartTime(oldValue.getSeriesStartTime()); |
| cacheValue.setSupportsAggregates(oldValue.isSupportsAggregates()); |
| cacheValue.setType(oldValue.getType()); |
| cacheValue.setIsWhitelisted(oldValue.isWhitelisted()); |
| } else if (oldValue.getSeriesStartTime() < cacheValue.getSeriesStartTime() && |
| cacheValue.getSeriesStartTime() != 0L && |
| cacheValue.isWhitelisted()) |
| { |
| LOG.info(String.format("Updating startTime for %s", key)); |
| cacheValue.setSeriesStartTime(oldValue.getSeriesStartTime()); |
| cacheValue.setIsPersisted(false); |
| } |
| } |
| } |
| } |
| |
| private boolean hasWildCard(String key) { |
| return StringUtils.isNotEmpty(key) && key.contains("%"); |
| } |
| |
| } |
