http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java index 371d5d2..23291bc 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java @@ -407,6 +407,23 @@ public class Configuration { private static final String DEFAULT_JDBC_POOL_ACQUISITION_RETRY_ATTEMPTS = "30"; private static final String DEFAULT_JDBC_POOL_ACQUISITION_RETRY_DELAY = "1000"; + // Timeline Metrics Cache settings + private static final String TIMELINE_METRICS_CACHE_DISABLE = "server.timeline.metrics.cache.disabled"; + private static final String TIMELINE_METRICS_CACHE_MAX_ENTRIES = "server.timeline.metrics.cache.max.entries"; + private static final String DEFAULT_TIMELINE_METRICS_CACHE_MAX_ENTRIES = "50"; + private static final String TIMELINE_METRICS_CACHE_TTL = "server.timeline.metrics.cache.entry.ttl.seconds"; + private static final String DEFAULT_TIMELINE_METRICS_CACHE_TTL = "3600"; + private static final String TIMELINE_METRICS_CACHE_IDLE_TIME = "server.timeline.metrics.cache.entry.idle.seconds"; + private static final String DEFAULT_TIMELINE_METRICS_CACHE_IDLE_TIME = "300"; + private static final String TIMELINE_METRICS_REQUEST_READ_TIMEOUT = "server.timeline.metrics.cache.read.timeout.millis"; + private static final String DEFAULT_TIMELINE_METRICS_REQUEST_READ_TIMEOUT = "10000"; + private static final String TIMELINE_METRICS_REQUEST_INTERVAL_READ_TIMEOUT = "server.timeline.metrics.cache.interval.read.timeout.millis"; + private static final String DEFAULT_TIMELINE_METRICS_REQUEST_INTERVAL_READ_TIMEOUT = "5000"; + private static final String TIMELINE_METRICS_REQUEST_CONNECT_TIMEOUT = "server.timeline.metrics.cache.connect.timeout.millis"; + private static final String DEFAULT_TIMELINE_METRICS_REQUEST_CONNECT_TIMEOUT = "5000"; + private static final String TIMELINE_METRICS_REQUEST_CATCHUP_INTERVAL = "server.timeline.metrics.cache.catchup.interval"; + private static final String DEFAULT_TIMELINE_METRICS_REQUEST_CATCHUP_INTERVAL = "300000"; + /** * The full path to the XML file that describes the different alert templates. */ @@ -679,6 +696,15 @@ public class Configuration { } /** + * Get the property value for the given key. + * + * @return the property value + */ + public String getProperty(String key, String defaultValue) { + return properties.getProperty(key, defaultValue); + } + + /** * Gets a copy of all of the configuration properties that back this * {@link Configuration} instance. * @@ -1001,7 +1027,7 @@ public class Configuration { */ public String getApiGzipMinSize() { return properties.getProperty(API_GZIP_MIN_COMPRESSION_SIZE_KEY, - API_GZIP_MIN_COMPRESSION_SIZE_DEFAULT); + API_GZIP_MIN_COMPRESSION_SIZE_DEFAULT); } /** @@ -1244,7 +1270,7 @@ public class Configuration { public int getConnectionMaxIdleTime() { return Integer.parseInt(properties.getProperty - (SERVER_CONNECTION_MAX_IDLE_TIME, String.valueOf("900000"))); + (SERVER_CONNECTION_MAX_IDLE_TIME, String.valueOf("900000"))); } /** @@ -1283,7 +1309,7 @@ public class Configuration { public int getOneWayAuthPort() { return Integer.parseInt(properties.getProperty(SRVR_ONE_WAY_SSL_PORT_KEY, - String.valueOf(SRVR_ONE_WAY_SSL_PORT_DEFAULT))); + String.valueOf(SRVR_ONE_WAY_SSL_PORT_DEFAULT))); } public int getTwoWayAuthPort() { @@ -1376,7 +1402,7 @@ public class Configuration { public Integer getRequestReadTimeout() { return Integer.parseInt(properties.getProperty(REQUEST_READ_TIMEOUT, - REQUEST_READ_TIMEOUT_DEFAULT)); + REQUEST_READ_TIMEOUT_DEFAULT)); } public Integer getRequestConnectTimeout() { @@ -1386,7 +1412,7 @@ public class Configuration { public String getExecutionSchedulerConnections() { return properties.getProperty(EXECUTION_SCHEDULER_CONNECTIONS, - DEFAULT_SCHEDULER_MAX_CONNECTIONS); + DEFAULT_SCHEDULER_MAX_CONNECTIONS); } public Long getExecutionSchedulerMisfireToleration() { @@ -1412,7 +1438,7 @@ public class Configuration { public String getCustomActionDefinitionPath() { return properties.getProperty(CUSTOM_ACTION_DEFINITION_KEY, - CUSTOM_ACTION_DEFINITION_DEF_VALUE); + CUSTOM_ACTION_DEFINITION_DEF_VALUE); } public int getAgentPackageParallelCommandsLimit() { @@ -1461,7 +1487,7 @@ public class Configuration { */ public int getClientThreadPoolSize() { return Integer.parseInt(properties.getProperty( - CLIENT_THREADPOOL_SIZE_KEY, String.valueOf(CLIENT_THREADPOOL_SIZE_DEFAULT))); + CLIENT_THREADPOOL_SIZE_KEY, String.valueOf(CLIENT_THREADPOOL_SIZE_DEFAULT))); } /** @@ -1499,7 +1525,7 @@ public class Configuration { */ public long getViewExtractionThreadPoolTimeout() { return Long.parseLong(properties.getProperty( - VIEW_EXTRACTION_THREADPOOL_TIMEOUT_KEY, String.valueOf(VIEW_EXTRACTION_THREADPOOL_TIMEOUT_DEFAULT))); + VIEW_EXTRACTION_THREADPOOL_TIMEOUT_KEY, String.valueOf(VIEW_EXTRACTION_THREADPOOL_TIMEOUT_DEFAULT))); } /** @@ -1512,7 +1538,7 @@ public class Configuration { */ public int getHttpSessionInactiveTimeout() { return Integer.parseInt(properties.getProperty( - SERVER_HTTP_SESSION_INACTIVE_TIMEOUT, + SERVER_HTTP_SESSION_INACTIVE_TIMEOUT, "1800")); } @@ -1531,7 +1557,7 @@ public class Configuration { */ public int getAlertEventPublisherPoolSize() { return Integer.parseInt(properties.getProperty( - ALERTS_EXECUTION_SCHEDULER_THREADS_KEY, ALERTS_EXECUTION_SCHEDULER_THREADS_DEFAULT)); + ALERTS_EXECUTION_SCHEDULER_THREADS_KEY, ALERTS_EXECUTION_SCHEDULER_THREADS_DEFAULT)); } /** @@ -1594,7 +1620,7 @@ public class Configuration { */ public int getKdcConnectionCheckTimeout() { return Integer.parseInt(properties.getProperty( - KDC_CONNECTION_CHECK_TIMEOUT_KEY, KDC_CONNECTION_CHECK_TIMEOUT_DEFAULT)); + KDC_CONNECTION_CHECK_TIMEOUT_KEY, KDC_CONNECTION_CHECK_TIMEOUT_DEFAULT)); } /** @@ -1779,4 +1805,73 @@ public class Configuration { } } + /** + * Max allowed entries in metrics cache. + */ + public int getMetricCacheMaxEntries() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_CACHE_MAX_ENTRIES, + DEFAULT_TIMELINE_METRICS_CACHE_MAX_ENTRIES)); + } + + /** + * Eviction time for entries in metrics cache. + */ + public int getMetricCacheTTLSeconds() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_CACHE_TTL, + DEFAULT_TIMELINE_METRICS_CACHE_TTL)); + } + + /** + * Max time to idle for entries in the cache. + */ + public int getMetricCacheIdleSeconds() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_CACHE_IDLE_TIME, + DEFAULT_TIMELINE_METRICS_CACHE_IDLE_TIME)); + } + + /** + * Separate timeout settings for metrics cache. + * @return milliseconds + */ + public int getMetricsRequestReadTimeoutMillis() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_REQUEST_READ_TIMEOUT, + DEFAULT_TIMELINE_METRICS_REQUEST_READ_TIMEOUT)); + } + + /** + * Separate timeout settings for metrics cache. + * Timeout on reads for update requests made for smaller time intervals. + * + * @return milliseconds + */ + public int getMetricsRequestIntervalReadTimeoutMillis() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_REQUEST_INTERVAL_READ_TIMEOUT, + DEFAULT_TIMELINE_METRICS_REQUEST_INTERVAL_READ_TIMEOUT)); + } + + /** + * Separate timeout settings for metrics cache. + * @return milliseconds + */ + public int getMetricsRequestConnectTimeoutMillis() { + return Integer.parseInt(properties.getProperty(TIMELINE_METRICS_REQUEST_CONNECT_TIMEOUT, + DEFAULT_TIMELINE_METRICS_REQUEST_CONNECT_TIMEOUT)); + } + + /** + * Diable metrics caching. + * @return true / false + */ + public boolean isMetricsCacheDisabled() { + return Boolean.parseBoolean(properties.getProperty(TIMELINE_METRICS_CACHE_DISABLE, "false")); + } + + /** + * Constant fudge factor subtracted from the cache update requests to + * account for unavailability of data on the trailing edge due to buffering. + */ + public Long getMetricRequestBufferTimeCatchupInterval() { + return Long.parseLong(properties.getProperty(TIMELINE_METRICS_REQUEST_CATCHUP_INTERVAL, + DEFAULT_TIMELINE_METRICS_REQUEST_CATCHUP_INTERVAL)); + } }
http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementController.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementController.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementController.java index 17b6d4a..0eef06c 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementController.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementController.java @@ -23,7 +23,9 @@ import org.apache.ambari.server.RoleCommand; import org.apache.ambari.server.actionmanager.ActionManager; import org.apache.ambari.server.agent.ExecutionCommand; import org.apache.ambari.server.api.services.AmbariMetaInfo; +import org.apache.ambari.server.configuration.Configuration; import org.apache.ambari.server.controller.internal.RequestStageContainer; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.metadata.RoleCommandOrder; import org.apache.ambari.server.scheduler.ExecutionScheduleManager; import org.apache.ambari.server.security.ldap.LdapBatchDto; @@ -782,5 +784,7 @@ public interface AmbariManagementController { * @return */ Set<StackConfigurationDependencyResponse> getStackConfigurationDependencies(Set<StackConfigurationDependencyRequest> requests) throws AmbariException; + + TimelineMetricCacheProvider getTimelineMetricCacheProvider(); } http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java index a7f206a..ef6fc58 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java @@ -89,6 +89,7 @@ import org.apache.ambari.server.controller.internal.RequestStageContainer; import org.apache.ambari.server.controller.internal.URLStreamProvider; import org.apache.ambari.server.controller.internal.WidgetLayoutResourceProvider; import org.apache.ambari.server.controller.internal.WidgetResourceProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.customactions.ActionDefinition; import org.apache.ambari.server.metadata.ActionMetadata; @@ -734,7 +735,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle handleGlobalsBackwardsCompability(request, propertiesAttributes); Config config = createConfig(cluster, request.getType(), request.getProperties(), - request.getVersionTag(), propertiesAttributes); + request.getVersionTag(), propertiesAttributes); return new ConfigurationResponse(cluster.getClusterName(), config); } @@ -782,7 +783,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle private Config createConfig(Cluster cluster, String type, Map<String, String> properties, String versionTag, Map<String, Map<String, String>> propertiesAttributes) { Config config = configFactory.createNew(cluster, type, - properties, propertiesAttributes); + properties, propertiesAttributes); if (!StringUtils.isEmpty(versionTag)) { config.setTag(versionTag); @@ -889,10 +890,10 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle String logDir = BASE_LOG_DIR + File.pathSeparator + requestId; Stage stage = stageFactory.createNew(requestId, logDir, - null == cluster ? null : cluster.getClusterName(), - null == cluster ? -1L : cluster.getClusterId(), - requestContext, clusterHostInfo, commandParamsStage, - hostParamsStage); + null == cluster ? null : cluster.getClusterName(), + null == cluster ? -1L : cluster.getClusterId(), + requestContext, clusterHostInfo, commandParamsStage, + hostParamsStage); stage.setStageId(id); return stage; } @@ -904,9 +905,9 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle if (LOG.isDebugEnabled()) { LOG.debug("Received a getClusters request" - + ", clusterName=" + request.getClusterName() - + ", clusterId=" + request.getClusterId() - + ", stackInfo=" + request.getStackVersion()); + + ", clusterName=" + request.getClusterName() + + ", clusterId=" + request.getClusterId() + + ", stackInfo=" + request.getStackVersion()); } Cluster singleCluster = null; @@ -2474,7 +2475,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle // Add attributes Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes(cluster, - ec.getConfigurationTags()); + ec.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); @@ -2606,7 +2607,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle boolean runSmokeTest, boolean reconfigureClients) throws AmbariException { RequestStageContainer request = addStages(null, cluster, requestProperties, requestParameters, changedServices, - changedComponents, changedHosts, ignoredHosts, runSmokeTest, reconfigureClients); + changedComponents, changedHosts, ignoredHosts, runSmokeTest, reconfigureClients); request.persist(); return request.getRequestStatusResponse(); @@ -3227,15 +3228,15 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle final Resource.Type level) throws AmbariException { Set<String> ignoredHosts = maintenanceStateHelper.filterHostsInMaintenanceState( - candidateHosts, new MaintenanceStateHelper.HostPredicate() { - @Override - public boolean shouldHostBeRemoved(final String hostname) - throws AmbariException { - Host host = clusters.getHost(hostname); - return !maintenanceStateHelper.isOperationAllowed( - host, cluster.getClusterId(), level); - } - } + candidateHosts, new MaintenanceStateHelper.HostPredicate() { + @Override + public boolean shouldHostBeRemoved(final String hostname) + throws AmbariException { + Host host = clusters.getHost(hostname); + return !maintenanceStateHelper.isOperationAllowed( + host, cluster.getClusterId(), level); + } + } ); LOG.debug("Ignoring hosts when selecting available hosts for action" + " due to maintenance state." + @@ -4188,7 +4189,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle String user, long createTime) { List<WidgetEntity> createdEntities = widgetDAO.findByName(clusterEntity.getClusterId(), layoutInfo.getWidgetName(), - user, layoutInfo.getDefaultSectionName()); + user, layoutInfo.getDefaultSectionName()); if (createdEntities == null || createdEntities.isEmpty()) { WidgetEntity widgetEntity = new WidgetEntity(); @@ -4329,4 +4330,8 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle } } + @Override + public TimelineMetricCacheProvider getTimelineMetricCacheProvider() { + return injector.getInstance(TimelineMetricCacheProvider.class); + } } http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java index 5644ca5..3c598db 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java @@ -63,6 +63,7 @@ import org.apache.ambari.server.controller.internal.StackDefinedPropertyProvider import org.apache.ambari.server.controller.internal.StackDependencyResourceProvider; import org.apache.ambari.server.controller.internal.UserPrivilegeResourceProvider; import org.apache.ambari.server.controller.internal.ViewPermissionResourceProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.utilities.DatabaseChecker; import org.apache.ambari.server.orm.GuiceJpaInitializer; import org.apache.ambari.server.orm.PersistenceType; @@ -649,7 +650,7 @@ public class AmbariServer { StageUtils.setGson(injector.getInstance(Gson.class)); StageUtils.setTopologyManager(injector.getInstance(TopologyManager.class)); WorkflowJsonService.setDBProperties( - injector.getInstance(Configuration.class)); + injector.getInstance(Configuration.class)); SecurityFilter.init(injector.getInstance(Configuration.class)); StackDefinedPropertyProvider.init(injector); AbstractControllerResourceProvider.init(injector.getInstance(ResourceProviderFactory.class)); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java index e048ec5..a40fae6 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java @@ -68,6 +68,8 @@ import org.apache.ambari.server.controller.internal.HostResourceProvider; import org.apache.ambari.server.controller.internal.MemberResourceProvider; import org.apache.ambari.server.controller.internal.RepositoryVersionResourceProvider; import org.apache.ambari.server.controller.internal.ServiceResourceProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheEntryFactory; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.ResourceProvider; import org.apache.ambari.server.controller.utilities.DatabaseChecker; import org.apache.ambari.server.notifications.DispatchFactory; @@ -329,6 +331,8 @@ public class ControllerModule extends AbstractModule { bind(ExecutionScheduler.class).to(ExecutionSchedulerImpl.class); bind(DBAccessor.class).to(DBAccessorImpl.class); bind(ViewInstanceHandlerList.class).to(AmbariHandlerList.class); + bind(TimelineMetricCacheProvider.class); + bind(TimelineMetricCacheEntryFactory.class); requestStaticInjection(ExecutionCommandWrapper.class); requestStaticInjection(DatabaseChecker.class); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractProviderModule.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractProviderModule.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractProviderModule.java index 380a0fe..6e3dcd5 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractProviderModule.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractProviderModule.java @@ -34,6 +34,8 @@ import org.apache.ambari.server.controller.metrics.MetricHostProvider; import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider; import org.apache.ambari.server.controller.metrics.MetricsReportPropertyProvider; import org.apache.ambari.server.controller.metrics.MetricsServiceProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheEntryFactory; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.NoSuchParentResourceException; import org.apache.ambari.server.controller.spi.NoSuchResourceException; import org.apache.ambari.server.controller.spi.Predicate; @@ -164,6 +166,9 @@ public abstract class AbstractProviderModule implements ProviderModule, @Inject AmbariManagementController managementController; + @Inject + TimelineMetricCacheProvider metricCacheProvider; + /** * The map of host components. */ @@ -200,6 +205,9 @@ public abstract class AbstractProviderModule implements ProviderModule, if (managementController == null) { managementController = AmbariServer.getController(); } + if (metricCacheProvider == null) { + metricCacheProvider = managementController.getTimelineMetricCacheProvider(); + } } @@ -884,7 +892,8 @@ public abstract class AbstractProviderModule implements ProviderModule, return MetricsReportPropertyProvider.createInstance( PropertyHelper.getMetricPropertyIds(type), streamProvider, - configuration, hostProvider, serviceProvider, clusterNamePropertyId); + configuration, metricCacheProvider, hostProvider, serviceProvider, + clusterNamePropertyId); } /** @@ -899,7 +908,7 @@ public abstract class AbstractProviderModule implements ProviderModule, String hostNamePropertyId) { return MetricsPropertyProvider.createInstance(type, PropertyHelper.getMetricPropertyIds(type), streamProvider, configuration, - hostProvider, serviceProvider, clusterNamePropertyId, + metricCacheProvider, hostProvider, serviceProvider, clusterNamePropertyId, hostNamePropertyId, null); } @@ -915,7 +924,8 @@ public abstract class AbstractProviderModule implements ProviderModule, String componentNamePropertyId) { return MetricsPropertyProvider.createInstance(type, PropertyHelper.getMetricPropertyIds(type), streamProvider, configuration, - hostProvider, serviceProvider, clusterNamePropertyId, null, + metricCacheProvider, hostProvider, serviceProvider, + clusterNamePropertyId, null, componentNamePropertyId); } @@ -934,8 +944,8 @@ public abstract class AbstractProviderModule implements ProviderModule, return MetricsPropertyProvider.createInstance(type, PropertyHelper.getMetricPropertyIds(type), streamProvider, configuration, - hostProvider, serviceProvider, clusterNamePropertyId, hostNamePropertyId, - componentNamePropertyId); + metricCacheProvider, hostProvider, serviceProvider, clusterNamePropertyId, + hostNamePropertyId, componentNamePropertyId); } @Override http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java index 3a6d30b..d0d597e 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java @@ -26,6 +26,7 @@ import org.apache.ambari.server.controller.jmx.JMXPropertyProvider; import org.apache.ambari.server.controller.metrics.MetricHostProvider; import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider; import org.apache.ambari.server.controller.metrics.MetricsServiceProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Predicate; import org.apache.ambari.server.controller.spi.PropertyProvider; import org.apache.ambari.server.controller.spi.Request; @@ -81,6 +82,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider { private final MetricHostProvider metricHostProvider; private final MetricsServiceProvider metricsServiceProvider; private MetricsService metricsService = MetricsService.GANGLIA; + private TimelineMetricCacheProvider cacheProvider; /** * PropertyHelper/AbstractPropertyProvider expect map of maps, @@ -126,6 +128,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider { this.streamProvider = streamProvider; defaultJmx = defaultJmxPropertyProvider; defaultGanglia = defaultGangliaPropertyProvider; + cacheProvider = injector.getInstance(TimelineMetricCacheProvider.class); } @@ -147,6 +150,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider { jmxStatePropertyId, defaultJmxPropertyProvider, defaultGangliaPropertyProvider); this.metricsService = metricsService; + cacheProvider = injector.getInstance(TimelineMetricCacheProvider.class); } @@ -201,7 +205,9 @@ public class StackDefinedPropertyProvider implements PropertyProvider { if (gangliaMap.size() > 0) { PropertyProvider propertyProvider = MetricsPropertyProvider.createInstance(type, gangliaMap, - streamProvider, sslConfig, metricHostProvider, + streamProvider, sslConfig, + cacheProvider, + metricHostProvider, metricsServiceProvider, clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/TemporalInfoImpl.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/TemporalInfoImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/TemporalInfoImpl.java index 2ffe984..4083152 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/TemporalInfoImpl.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/TemporalInfoImpl.java @@ -20,6 +20,8 @@ package org.apache.ambari.server.controller.internal; import org.apache.ambari.server.controller.spi.TemporalInfo; +import java.util.Date; + /** * Temporal query data. */ @@ -74,6 +76,15 @@ public class TemporalInfoImpl implements TemporalInfo { } @Override + public String toString() { + return "TemporalInfoImpl{" + + "m_startTime = " + new Date(getStartTimeMillis()) + + ", m_endTime = " + new Date(getEndTimeMillis()) + + ", m_step = " + m_step + + '}'; + } + + @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; @@ -82,7 +93,6 @@ public class TemporalInfoImpl implements TemporalInfo { return m_endTime == that.m_endTime && m_startTime == that.m_startTime && m_step == that.m_step; - } @Override http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPaddingMethod.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPaddingMethod.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPaddingMethod.java index 32d214d..522c0bd 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPaddingMethod.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPaddingMethod.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics; import org.apache.ambari.server.controller.spi.TemporalInfo; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; import java.util.Iterator; +import java.util.Map; import java.util.TreeMap; public class MetricsPaddingMethod { @@ -51,8 +52,15 @@ public class MetricsPaddingMethod { return; } - // TODO: JSON dser returns LinkedHashMap that is not Navigable - TreeMap<Long, Double> values = new TreeMap<Long, Double>(metric.getMetricValues()); + TreeMap<Long, Double> values; + Map<Long, Double> metricValuesMap = metric.getMetricValues(); + if (metricValuesMap instanceof TreeMap) { + values = (TreeMap<Long, Double>) metricValuesMap; + } + else { + // JSON dser returns LinkedHashMap that is not Navigable + values = new TreeMap<Long, Double>(metricValuesMap); + } long dataInterval = getTimelineMetricInterval(values); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java index 9fa9ca4..ad35444 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.internal.AbstractPropertyProvider; import org.apache.ambari.server.controller.internal.PropertyInfo; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Predicate; import org.apache.ambari.server.controller.spi.Request; import org.apache.ambari.server.controller.spi.Resource; @@ -81,6 +82,7 @@ public abstract class MetricsPropertyProvider extends AbstractPropertyProvider { Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, MetricsServiceProvider serviceProvider, String clusterNamePropertyId, @@ -93,6 +95,7 @@ public abstract class MetricsPropertyProvider extends AbstractPropertyProvider { componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, serviceProvider, clusterNamePropertyId, http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProviderProxy.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProviderProxy.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProviderProxy.java index 57a8e7d..fb74fab 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProviderProxy.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProviderProxy.java @@ -28,6 +28,7 @@ import org.apache.ambari.server.controller.metrics.timeline.AMSComponentProperty import org.apache.ambari.server.controller.metrics.timeline.AMSHostComponentPropertyProvider; import org.apache.ambari.server.controller.metrics.timeline.AMSHostPropertyProvider; import org.apache.ambari.server.controller.metrics.timeline.AMSPropertyProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Predicate; import org.apache.ambari.server.controller.spi.Request; import org.apache.ambari.server.controller.spi.Resource; @@ -46,12 +47,14 @@ public class MetricsPropertyProviderProxy extends AbstractPropertyProvider { private final MetricsServiceProvider metricsServiceProvider; private AMSPropertyProvider amsPropertyProvider; private GangliaPropertyProvider gangliaPropertyProvider; + private TimelineMetricCacheProvider cacheProvider; public MetricsPropertyProviderProxy( InternalType type, Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, MetricsServiceProvider serviceProvider, String clusterNamePropertyId, @@ -60,6 +63,7 @@ public class MetricsPropertyProviderProxy extends AbstractPropertyProvider { super(componentPropertyInfoMap); this.metricsServiceProvider = serviceProvider; + this.cacheProvider = cacheProvider; switch (type) { case Host: @@ -108,6 +112,7 @@ public class MetricsPropertyProviderProxy extends AbstractPropertyProvider { this.amsPropertyProvider = new AMSHostPropertyProvider(componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, hostNamePropertyId); @@ -132,6 +137,7 @@ public class MetricsPropertyProviderProxy extends AbstractPropertyProvider { componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, hostNamePropertyId, @@ -158,6 +164,7 @@ public class MetricsPropertyProviderProxy extends AbstractPropertyProvider { componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, componentNamePropertyId); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java index 5399436..54857cc 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.internal.AbstractPropertyProvider; import org.apache.ambari.server.controller.internal.PropertyInfo; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.utilities.StreamProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -63,6 +64,7 @@ public abstract class MetricsReportPropertyProvider extends AbstractPropertyProv Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, MetricsServiceProvider serviceProvider, String clusterNamePropertyId) { @@ -70,6 +72,7 @@ public abstract class MetricsReportPropertyProvider extends AbstractPropertyProv return new MetricsReportPropertyProviderProxy(componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, serviceProvider, clusterNamePropertyId); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProviderProxy.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProviderProxy.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProviderProxy.java index a92cb37..9b23686 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProviderProxy.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProviderProxy.java @@ -22,6 +22,7 @@ import org.apache.ambari.server.controller.internal.AbstractPropertyProvider; import org.apache.ambari.server.controller.internal.PropertyInfo; import org.apache.ambari.server.controller.metrics.ganglia.GangliaReportPropertyProvider; import org.apache.ambari.server.controller.metrics.timeline.AMSReportPropertyProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Predicate; import org.apache.ambari.server.controller.spi.Request; import org.apache.ambari.server.controller.spi.Resource; @@ -39,11 +40,13 @@ public class MetricsReportPropertyProviderProxy extends AbstractPropertyProvider private MetricsReportPropertyProvider amsMetricsReportProvider; private MetricsReportPropertyProvider gangliaMetricsReportProvider; private final MetricsServiceProvider metricsServiceProvider; + private TimelineMetricCacheProvider cacheProvider; public MetricsReportPropertyProviderProxy( Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, MetricsServiceProvider serviceProvider, String clusterNamePropertyId) { @@ -51,6 +54,7 @@ public class MetricsReportPropertyProviderProxy extends AbstractPropertyProvider super(componentPropertyInfoMap); this.metricsServiceProvider = serviceProvider; + this.cacheProvider = cacheProvider; createReportPropertyProviders(componentPropertyInfoMap, streamProvider, @@ -69,6 +73,7 @@ public class MetricsReportPropertyProviderProxy extends AbstractPropertyProvider componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSComponentPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSComponentPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSComponentPropertyProvider.java index d5f415a..cc0219c 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSComponentPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSComponentPropertyProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics.timeline; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.internal.PropertyInfo; import org.apache.ambari.server.controller.metrics.MetricHostProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.utilities.StreamProvider; @@ -30,12 +31,13 @@ public class AMSComponentPropertyProvider extends AMSPropertyProvider { public AMSComponentPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, String clusterNamePropertyId, String componentNamePropertyId) { - super(componentPropertyInfoMap, streamProvider, configuration, hostProvider, - clusterNamePropertyId, null, componentNamePropertyId); + super(componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, null, componentNamePropertyId); } @Override http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostComponentPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostComponentPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostComponentPropertyProvider.java index 8ee2acb..0e71049 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostComponentPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostComponentPropertyProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics.timeline; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.internal.PropertyInfo; import org.apache.ambari.server.controller.metrics.MetricHostProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.utilities.StreamProvider; import java.util.Map; @@ -29,13 +30,15 @@ public class AMSHostComponentPropertyProvider extends AMSPropertyProvider { public AMSHostComponentPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, String clusterNamePropertyId, String hostNamePropertyId, String componentNamePropertyId) { - super(componentPropertyInfoMap, streamProvider, configuration, hostProvider, - clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId); + super(componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, hostNamePropertyId, + componentNamePropertyId); } @Override http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostPropertyProvider.java index ca9d685..85e016f 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSHostPropertyProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.metrics.timeline; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.internal.PropertyInfo; import org.apache.ambari.server.controller.metrics.MetricHostProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.utilities.StreamProvider; @@ -30,12 +31,13 @@ public class AMSHostPropertyProvider extends AMSPropertyProvider { public AMSHostPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, String clusterNamePropertyId, String hostNamePropertyId) { - super(componentPropertyInfoMap, streamProvider, configuration, hostProvider, - clusterNamePropertyId, hostNamePropertyId, null); + super(componentPropertyInfoMap, streamProvider, configuration, + cacheProvider, hostProvider, clusterNamePropertyId, hostNamePropertyId, null); } @Override http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java index 6667134..55b7349 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java @@ -17,6 +17,7 @@ */ package org.apache.ambari.server.controller.metrics.timeline; +import com.google.common.collect.Sets; import org.apache.ambari.server.api.services.AmbariMetaInfo; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; import org.apache.ambari.server.controller.AmbariManagementController; @@ -24,6 +25,9 @@ import org.apache.ambari.server.controller.AmbariServer; import org.apache.ambari.server.controller.internal.PropertyInfo; import org.apache.ambari.server.controller.metrics.MetricHostProvider; import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineAppMetricCacheKey; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCache; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Request; import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.spi.SystemException; @@ -35,13 +39,8 @@ import org.apache.commons.lang.StringUtils; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics; import org.apache.http.client.utils.URIBuilder; -import org.codehaus.jackson.map.AnnotationIntrospector; -import org.codehaus.jackson.map.ObjectMapper; -import org.codehaus.jackson.map.ObjectReader; -import org.codehaus.jackson.xc.JaxbAnnotationIntrospector; -import java.io.BufferedReader; + import java.io.IOException; -import java.io.InputStreamReader; import java.net.SocketTimeoutException; import java.util.Collection; import java.util.Collections; @@ -51,33 +50,28 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; import java.util.regex.Pattern; + import static org.apache.ambari.server.Role.HBASE_MASTER; import static org.apache.ambari.server.Role.HBASE_REGIONSERVER; import static org.apache.ambari.server.Role.METRICS_COLLECTOR; import static org.apache.ambari.server.controller.metrics.MetricsPaddingMethod.ZERO_PADDING_PARAM; import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.TIMELINE_METRICS; -import static org.codehaus.jackson.map.annotate.JsonSerialize.Inclusion; public abstract class AMSPropertyProvider extends MetricsPropertyProvider { - private static ObjectMapper mapper; - private final static ObjectReader timelineObjectReader; private static final String METRIC_REGEXP_PATTERN = "\\([^)]*\\)"; private static final int COLLECTOR_DEFAULT_PORT = 6188; - - static { - mapper = new ObjectMapper(); - AnnotationIntrospector introspector = new JaxbAnnotationIntrospector(); - mapper.setAnnotationIntrospector(introspector); - //noinspection deprecation - mapper.getSerializationConfig().setSerializationInclusion(Inclusion.NON_NULL); - timelineObjectReader = mapper.reader(TimelineMetrics.class); - } + private final TimelineMetricCache metricCache; + private static AtomicInteger printSkipPopulateMsgHostCounter = new AtomicInteger(0); + private static AtomicInteger printSkipPopulateMsgHostCompCounter = new AtomicInteger(0); public AMSPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, String clusterNamePropertyId, String hostNamePropertyId, @@ -86,6 +80,8 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { super(componentPropertyInfoMap, streamProvider, configuration, hostProvider, clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId); + + this.metricCache = cacheProvider.getTimelineMetricsCache(); } protected String getOverridenComponentName(Resource resource) { @@ -121,6 +117,9 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { private final Map<String, Set<Resource>> resources = new HashMap<String, Set<Resource>>(); private final Map<String, Set<String>> metrics = new HashMap<String, Set<String>>(); private final URIBuilder uriBuilder; + Set<String> resolvedMetricsParams; + MetricsRequestHelper requestHelper = new MetricsRequestHelper(streamProvider); + // Metrics with amsHostMetric = true // Basically a host metric to be returned for a hostcomponent private final Set<String> hostComponentHostMetrics = new HashSet<String>(); @@ -162,45 +161,22 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { } } - private TimelineMetrics getTimelineMetricsForSpec(String spec) { - TimelineMetrics timelineMetrics = null; - - LOG.debug("Metrics request url = " + spec); - BufferedReader reader = null; - try { - reader = new BufferedReader(new InputStreamReader(streamProvider.readFrom(spec))); - timelineMetrics = timelineObjectReader.readValue(reader); - LOG.debug("Timeline metrics response => " + timelineMetrics); - - } catch (IOException io) { - String errorMsg = "Error getting timeline metrics."; - if (LOG.isDebugEnabled()) { - LOG.error(errorMsg, io); - } else { - if (io instanceof SocketTimeoutException) { - errorMsg += " Can not connect to collector, socket error."; - } - LOG.error(errorMsg); - } - } finally { - if (reader != null) { - try { - reader.close(); - } catch (IOException e) { - if (LOG.isWarnEnabled()) { - if (LOG.isDebugEnabled()) { - LOG.warn("Unable to close http input stream : spec=" + spec, e); - } else { - LOG.warn("Unable to close http input stream : spec=" + spec); - } - } - } - } + private TimelineMetrics getTimelineMetricsFromCache(TimelineAppMetricCacheKey metricCacheKey, + String componentName) throws IOException { + // Cache only the component level metrics + // No point in time metrics are cached + if (metricCache != null + && !StringUtils.isEmpty(componentName) + && !componentName.equalsIgnoreCase("HOST") + && metricCacheKey.getTemporalInfo() != null) { + return metricCache.getAppTimelineMetricsFromCache(metricCacheKey); } - return timelineMetrics; + return requestHelper.fetchTimelineMetrics(metricCacheKey.getSpec()); } + + /** * Populate the associated resources by making a call to the Metrics * service. @@ -208,6 +184,7 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { * @return a collection of populated resources * @throws SystemException if unable to populate the resources */ + @SuppressWarnings("unchecked") public Collection<Resource> populateResources() throws SystemException { // No open ended query support. if (temporalInfo != null && (temporalInfo.getStartTime() == null @@ -219,27 +196,62 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { String componentName = resourceEntry.getKey(); Set<Resource> resourceSet = resourceEntry.getValue(); - TimelineMetrics timelineMetrics; - // Allow for multiple requests since host metrics for a - // hostcomponent need the HOST appId - if (hostComponentHostMetrics.isEmpty()) { //HOST - String spec = getSpec(componentName); - timelineMetrics = getTimelineMetricsForSpec(spec); - } else { - Set<String> specs = getSpecsForHostComponentMetrics(componentName); - timelineMetrics = new TimelineMetrics(); - for (String spec : specs) { - if (!StringUtils.isEmpty(spec)) { - TimelineMetrics metrics = getTimelineMetricsForSpec(spec); - if (metrics != null) { - timelineMetrics.getMetrics().addAll(metrics.getMetrics()); - } - } + TimelineMetrics timelineMetrics = new TimelineMetrics(); + + Set<String> nonHostComponentMetrics = new HashSet<String>(metrics.keySet()); + nonHostComponentMetrics.removeAll(hostComponentHostMetrics); + String hostnames = getHostnames(resources.get(componentName)); + + // Allow for multiple requests since host metrics for a + // hostcomponent need the HOST appId + if (!hostComponentHostMetrics.isEmpty()) { + String hostComponentHostMetricParams = getSetString(processRegexps(hostComponentHostMetrics), -1); + setQueryParams(hostComponentHostMetricParams, hostnames, true, componentName); + TimelineMetrics metricsResponse = null; + try { + metricsResponse = getTimelineMetricsFromCache( + getTimelineAppMetricCacheKey(hostComponentHostMetrics, + componentName, uriBuilder.toString()), componentName); + } catch (IOException e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Caught exception fetching metric data.", e); } + // Skip further queries to preempt long calls due to timeout + if (e instanceof SocketTimeoutException) { + return Collections.emptySet(); + } + } + if (metricsResponse != null) { + timelineMetrics.getMetrics().addAll(metricsResponse.getMetrics()); } + } + + if (!nonHostComponentMetrics.isEmpty()) { + String nonHostComponentHostMetricParams = getSetString(processRegexps(nonHostComponentMetrics), -1); + setQueryParams(nonHostComponentHostMetricParams, hostnames, false, componentName); + TimelineMetrics metricsResponse = null; + try { + metricsResponse = getTimelineMetricsFromCache( + getTimelineAppMetricCacheKey(nonHostComponentMetrics, + componentName, uriBuilder.toString()), componentName); + } catch (IOException e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Caught exception fetching metric data.", e); + } + // Skip further queries to preempt long calls due to timeout + if (e instanceof SocketTimeoutException) { + return Collections.emptySet(); + } + } + if (metricsResponse != null) { + timelineMetrics.getMetrics().addAll(metricsResponse.getMetrics()); + } + } + Map<String, Set<TimelineMetric>> metricsMap = new HashMap<String, Set<TimelineMetric>>(); Set<String> patterns = createPatterns(metrics.keySet()); - if (timelineMetrics != null) { + + if (!timelineMetrics.getMetrics().isEmpty()) { for (TimelineMetric metric : timelineMetrics.getMetrics()) { if (metric.getMetricName() != null && metric.getMetricValues() != null @@ -255,9 +267,11 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { String hostname = getHostName(resource); if (metricsMap.containsKey(hostname)) { for (TimelineMetric metric : metricsMap.get(hostname)) { - // Pad zeros or nulls if needed - metricsPaddingMethod.applyPaddingStrategy(metric, temporalInfo); - populateResource(resource, metric, temporalInfo); + // Pad zeros or nulls if needed to a clone so we do not cache + // padded values + TimelineMetric timelineMetricClone = new TimelineMetric(metric); + metricsPaddingMethod.applyPaddingStrategy(timelineMetricClone, temporalInfo); + populateResource(resource, timelineMetricClone, temporalInfo); } } } @@ -267,38 +281,14 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { return Collections.emptySet(); } - /** - * Return separate specs for : host component metrics and host component - * host metrics. - * @return @Set Urls - */ - private Set<String> getSpecsForHostComponentMetrics(String componentName) { - Set<String> nonHostComponentMetrics = new HashSet<String>(metrics.keySet()); - nonHostComponentMetrics.removeAll(hostComponentHostMetrics); - - Set<String> specs = new HashSet<String>(); - String hostnames = getHostnames(resources.get(componentName)); - if (!hostComponentHostMetrics.isEmpty()) { - String hostComponentHostMetricParams = getSetString(processRegexps(hostComponentHostMetrics), -1); - setQueryParams(hostComponentHostMetricParams, hostnames, true, componentName); - specs.add(uriBuilder.toString()); - } - - if (!nonHostComponentMetrics.isEmpty()) { - String nonHostComponentHostMetricParams = getSetString(processRegexps(nonHostComponentMetrics), -1); - setQueryParams(nonHostComponentHostMetricParams, hostnames, false, componentName); - specs.add(uriBuilder.toString()); - } - return specs; - } - - private void setQueryParams(String metricsParam, - String hostname, boolean isHostMetric, String componentName) { + private void setQueryParams(String metricsParam, String hostname, + boolean isHostMetric, String componentName) { // Reuse uriBuilder uriBuilder.removeQuery(); if (metricsParam.length() > 0) { uriBuilder.setParameter("metricNames", metricsParam); + resolvedMetricsParams = Sets.newHashSet(metricsParam.split(",")); } if (hostname != null && !hostname.isEmpty()) { @@ -309,7 +299,7 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { uriBuilder.setParameter("appId", "HOST"); } else { if (componentName != null && !componentName.isEmpty() - && !componentName.equalsIgnoreCase("HOST")) { + && !componentName.equalsIgnoreCase("HOST")) { StackId stackId; try { AmbariManagementController managementController = AmbariServer.getController(); @@ -344,14 +334,6 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { } } - private String getSpec(String componentName) { - String metricsParam = getSetString(processRegexps(metrics.keySet()), -1); - String hostnames = getHostnames(resources.get(componentName)); - setQueryParams(metricsParam, hostnames, false, componentName); - - return uriBuilder.toString(); - } - private Set<String> createPatterns(Set<String> rawNames) { Pattern pattern = Pattern.compile(METRIC_REGEXP_PATTERN); Set<String> result = new HashSet<String>(); @@ -432,6 +414,22 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { } } } + + // Called when host component metrics are present + private TimelineAppMetricCacheKey getTimelineAppMetricCacheKey(Set<String> metrics, + String componentName, String spec) { + + TimelineAppMetricCacheKey metricCacheKey = + new TimelineAppMetricCacheKey(metrics, componentName, temporalInfo); + + // Set Uri on the cache key so the only job of the cache update is + // tweaking the params. Note: Passing UriBuilder reference is unsafe + // due to reuse. Also, the Uri can only be constructed with a resource + // request which ties it to the cluster. + metricCacheKey.setSpec(spec); + + return metricCacheKey; + } } private String getHostnames(Set<Resource> resources) { @@ -496,6 +494,14 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { return super.getComponentMetrics(); } + /** + * Return a set of @MetricsRequest object for each cluster. + * + * @param resources Set of resources asked to populate + * @param request Original Request object used to check properties + * @param ids Property ids to populate on the resource + * @throws SystemException + */ private Map<String, Map<TemporalInfo, MetricsRequest>> getMetricsRequests( Set<Resource> resources, Request request, Set<String> ids) throws SystemException { @@ -515,17 +521,31 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { // Check liveliness of host if (!hostProvider.isCollectorHostLive(clusterName, TIMELINE_METRICS)) { - LOG.info("METRICS_COLLECTOR host is not live. Skip populating " + - "resources with metrics."); + if (printSkipPopulateMsgHostCounter.getAndIncrement() == 0) { + LOG.info("METRICS_COLLECTOR host is not live. Skip populating " + + "resources with metrics, next message will be logged after 1000 " + + "attempts."); + } else { + printSkipPopulateMsgHostCounter.compareAndSet(1000, 0); + } continue; } + // reset + printSkipPopulateMsgHostCounter.set(0); // Check liveliness of Collector if (!hostProvider.isCollectorComponentLive(clusterName, TIMELINE_METRICS)) { - LOG.info("METRICS_COLLECTOR is not live. Skip populating resources" + - " with metrics."); + if (printSkipPopulateMsgHostCompCounter.getAndIncrement() == 0) { + LOG.info("METRICS_COLLECTOR is not live. Skip populating resources " + + "with metrics., next message will be logged after 1000 " + + "attempts."); + } else { + printSkipPopulateMsgHostCompCounter.compareAndSet(1000, 0); + } continue; } + // reset + printSkipPopulateMsgHostCompCounter.set(0); Map<TemporalInfo, MetricsRequest> requests = requestMap.get(clusterName); if (requests == null) { @@ -568,7 +588,7 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider { metricsRequest = new MetricsRequest(temporalInfo, getAMSUriBuilder(collectorHostName, collectorPort != null ? Integer.parseInt(collectorPort) : COLLECTOR_DEFAULT_PORT), - (String) resource.getPropertyValue(clusterNamePropertyId)); + (String) resource.getPropertyValue(clusterNamePropertyId)); requests.put(temporalInfo, metricsRequest); } metricsRequest.putResource(getComponentName(resource), resource); http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSReportPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSReportPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSReportPropertyProvider.java index a095206..0605123 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSReportPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSReportPropertyProvider.java @@ -23,6 +23,9 @@ import org.apache.ambari.server.controller.metrics.MetricHostProvider; import org.apache.ambari.server.controller.metrics.MetricsPaddingMethod; import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider; import org.apache.ambari.server.controller.metrics.MetricsReportPropertyProvider; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineAppMetricCacheKey; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCache; +import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider; import org.apache.ambari.server.controller.spi.Predicate; import org.apache.ambari.server.controller.spi.Request; import org.apache.ambari.server.controller.spi.Resource; @@ -33,49 +36,37 @@ import org.apache.ambari.server.controller.utilities.StreamProvider; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics; import org.apache.http.client.utils.URIBuilder; -import org.codehaus.jackson.map.AnnotationIntrospector; -import org.codehaus.jackson.map.ObjectMapper; -import org.codehaus.jackson.map.ObjectReader; -import org.codehaus.jackson.map.annotate.JsonSerialize; -import org.codehaus.jackson.xc.JaxbAnnotationIntrospector; -import java.io.BufferedReader; import java.io.IOException; -import java.io.InputStreamReader; -import java.net.SocketTimeoutException; -import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; -import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.ambari.server.controller.metrics.MetricsPaddingMethod.ZERO_PADDING_PARAM; import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.TIMELINE_METRICS; -import static org.apache.ambari.server.controller.utilities.PropertyHelper.updateMetricsWithAggregateFunctionSupport; public class AMSReportPropertyProvider extends MetricsReportPropertyProvider { - private static ObjectMapper mapper; - private final static ObjectReader timelineObjectReader; private MetricsPaddingMethod metricsPaddingMethod; - - static { - mapper = new ObjectMapper(); - AnnotationIntrospector introspector = new JaxbAnnotationIntrospector(); - mapper.setAnnotationIntrospector(introspector); - //noinspection deprecation - mapper.getSerializationConfig().setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL); - timelineObjectReader = mapper.reader(TimelineMetrics.class); - } + private final TimelineMetricCache metricCache; + MetricsRequestHelper requestHelper; + private static AtomicInteger printSkipPopulateMsgHostCounter = new AtomicInteger(0); + private static AtomicInteger printSkipPopulateMsgHostCompCounter = new AtomicInteger(0); public AMSReportPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap, StreamProvider streamProvider, ComponentSSLConfiguration configuration, + TimelineMetricCacheProvider cacheProvider, MetricHostProvider hostProvider, String clusterNamePropertyId) { super(componentPropertyInfoMap, streamProvider, configuration, hostProvider, clusterNamePropertyId); + + this.metricCache = cacheProvider.getTimelineMetricsCache(); + this.requestHelper = new MetricsRequestHelper(streamProvider); } /** @@ -144,17 +135,33 @@ public class AMSReportPropertyProvider extends MetricsReportPropertyProvider { // Check liveliness of host if (!hostProvider.isCollectorHostLive(clusterName, TIMELINE_METRICS)) { - LOG.info("METRICS_COLLECTOR host is not live. Skip populating " + - "resources with metrics."); + if (printSkipPopulateMsgHostCounter.getAndIncrement() == 0) { + LOG.info("METRICS_COLLECTOR host is not live. Skip populating " + + "resources with metrics, next message will be logged after 1000 " + + "attempts."); + } else { + printSkipPopulateMsgHostCounter.compareAndSet(1000, 0); + } + return true; } + // reset + printSkipPopulateMsgHostCompCounter.set(0); // Check liveliness of Collector if (!hostProvider.isCollectorComponentLive(clusterName, TIMELINE_METRICS)) { - LOG.info("METRICS_COLLECTOR is not live. Skip populating resources" + - " with metrics."); + if (printSkipPopulateMsgHostCompCounter.getAndIncrement() == 0) { + LOG.info("METRICS_COLLECTOR is not live. Skip populating resources" + + " with metrics, next message will be logged after 1000 " + + "attempts."); + } else { + printSkipPopulateMsgHostCompCounter.compareAndSet(1000, 0); + } + return true; } + // reset + printSkipPopulateMsgHostCompCounter.set(0); setProperties(resource, clusterName, request, getRequestPropertyIds(request, predicate)); @@ -191,48 +198,34 @@ public class AMSReportPropertyProvider extends MetricsReportPropertyProvider { uriBuilder.setParameter("endTime", String.valueOf(endTime)); } - BufferedReader reader = null; - String spec = uriBuilder.toString(); - try { - LOG.debug("Metrics request url =" + spec); - reader = new BufferedReader(new InputStreamReader(streamProvider.readFrom(spec))); + TimelineAppMetricCacheKey metricCacheKey = + new TimelineAppMetricCacheKey(propertyIdMap.keySet(), "HOST", temporalInfo); + + metricCacheKey.setSpec(uriBuilder.toString()); - TimelineMetrics timelineMetrics = timelineObjectReader.readValue(reader); - LOG.debug("Timeline metrics response => " + timelineMetrics); + // Self populating cache updates itself on every get with latest results + TimelineMetrics timelineMetrics; + if (metricCache != null && metricCacheKey.getTemporalInfo() != null) { + timelineMetrics = metricCache.getAppTimelineMetricsFromCache(metricCacheKey); + } else { + try { + timelineMetrics = requestHelper.fetchTimelineMetrics(uriBuilder.toString()); + } catch (IOException e) { + timelineMetrics = null; + } + } + if (timelineMetrics != null) { for (TimelineMetric metric : timelineMetrics.getMetrics()) { if (metric.getMetricName() != null && metric.getMetricValues() != null) { - // Pad zeros or nulls if needed - metricsPaddingMethod.applyPaddingStrategy(metric, temporalInfo); + // Pad zeros or nulls if needed to a clone so we do not cache + // padded values + TimelineMetric timelineMetricClone = new TimelineMetric(metric); + metricsPaddingMethod.applyPaddingStrategy(timelineMetricClone, temporalInfo); String propertyId = propertyIdMap.get(metric.getMetricName()); if (propertyId != null) { - resource.setProperty(propertyId, getValue(metric, temporalInfo)); - } - } - } - - } catch (IOException io) { - String errorMsg = "Error getting timeline metrics."; - if (LOG.isDebugEnabled()) { - LOG.error(errorMsg, io); - } else { - if (io instanceof SocketTimeoutException) { - errorMsg += " Can not connect to collector, socket error."; - } - LOG.error(errorMsg); - } - } finally { - if (reader != null) { - try { - reader.close(); - } catch (IOException e) { - if (LOG.isWarnEnabled()) { - if (LOG.isDebugEnabled()) { - LOG.warn("Unable to close http input steam : spec=" + spec, e); - } else { - LOG.warn("Unable to close http input steam : spec=" + spec); - } + resource.setProperty(propertyId, getValue(timelineMetricClone, temporalInfo)); } } } http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/MetricsRequestHelper.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/MetricsRequestHelper.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/MetricsRequestHelper.java new file mode 100644 index 0000000..ca20e54 --- /dev/null +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/MetricsRequestHelper.java @@ -0,0 +1,108 @@ +/** + * 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.controller.metrics.timeline; + +import org.apache.ambari.server.controller.utilities.StreamProvider; +import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; +import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics; +import org.codehaus.jackson.map.AnnotationIntrospector; +import org.codehaus.jackson.map.ObjectMapper; +import org.codehaus.jackson.map.ObjectReader; +import org.codehaus.jackson.map.annotate.JsonSerialize; +import org.codehaus.jackson.xc.JaxbAnnotationIntrospector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.SocketTimeoutException; +import java.util.Date; +import java.util.Map; + +/** + * Helper class to call AMS backend that is utilized by @AMSPropertyProvider + * and @AMSReportPropertyProvider as well as @TimelineMetricCacheEntryFactory + */ +public class MetricsRequestHelper { + private final static Logger LOG = LoggerFactory.getLogger(MetricsRequestHelper.class); + private final static ObjectMapper mapper; + private final static ObjectReader timelineObjectReader; + private final StreamProvider streamProvider; + + static { + mapper = new ObjectMapper(); + AnnotationIntrospector introspector = new JaxbAnnotationIntrospector(); + mapper.setAnnotationIntrospector(introspector); + //noinspection deprecation + mapper.getSerializationConfig().setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL); + timelineObjectReader = mapper.reader(TimelineMetrics.class); + } + + public MetricsRequestHelper(StreamProvider streamProvider) { + this.streamProvider = streamProvider; + } + + public TimelineMetrics fetchTimelineMetrics(String spec) throws IOException { + LOG.debug("Metrics request url = " + spec); + BufferedReader reader = null; + TimelineMetrics timelineMetrics = null; + try { + reader = new BufferedReader(new InputStreamReader(streamProvider.readFrom(spec))); + timelineMetrics = timelineObjectReader.readValue(reader); + if (LOG.isTraceEnabled()) { + for (TimelineMetric metric : timelineMetrics.getMetrics()) { + LOG.trace("metric: " + metric.getMetricName() + + ", size = " + metric.getMetricValues().size() + + ", host = " + metric.getHostName() + + ", app = " + metric.getAppId() + + ", instance = " + metric.getInstanceId() + + ", time = " + metric.getTimestamp() + + ", startTime = " + new Date(metric.getStartTime())); + } + } + } catch (IOException io) { + String errorMsg = "Error getting timeline metrics."; + if (LOG.isDebugEnabled()) { + LOG.error(errorMsg, io); + } + + if (io instanceof SocketTimeoutException) { + errorMsg += " Can not connect to collector, socket error."; + LOG.error(errorMsg); + throw io; + } + + } finally { + if (reader != null) { + try { + reader.close(); + } catch (IOException e) { + if (LOG.isWarnEnabled()) { + if (LOG.isDebugEnabled()) { + LOG.warn("Unable to close http input stream : spec=" + spec, e); + } else { + LOG.warn("Unable to close http input stream : spec=" + spec); + } + } + } + } + } + return timelineMetrics; + } +} http://git-wip-us.apache.org/repos/asf/ambari/blob/02fd9a79/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineAppMetricCacheKey.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineAppMetricCacheKey.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineAppMetricCacheKey.java new file mode 100644 index 0000000..76bc73b --- /dev/null +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineAppMetricCacheKey.java @@ -0,0 +1,119 @@ +/** + * 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.controller.metrics.timeline.cache; + +import org.apache.ambari.server.controller.spi.TemporalInfo; +import org.apache.http.client.utils.URIBuilder; + +import java.util.Set; + +/** + * Cache contents represent metrics for an App / Cluster. + * This is designed to work on the premise that a client makes same requests + * over and over, so this caching strategy allows us to send a http request + * for multiple metrics in the same query and ensure parallelization on the + * metrics backend side as well. + */ +public class TimelineAppMetricCacheKey { + private final Set<String> metricNames; + private final String appId; + private String spec; + private TemporalInfo temporalInfo; + + public TimelineAppMetricCacheKey(Set<String> metricNames, String appId, + TemporalInfo temporalInfo) { + this.metricNames = metricNames; + this.appId = appId; + this.temporalInfo = temporalInfo; + } + + public Set<String> getMetricNames() { + return metricNames; + } + + /** + * Temporal info is used to calculate the next query window, + * it does not contribute to the key behavior. + * @return @TemporalInfo + */ + public TemporalInfo getTemporalInfo() { + return temporalInfo; + } + + /** + * Set temporalInfo to new query window each time. + * @param temporalInfo @TemporalInfo + */ + public void setTemporalInfo(TemporalInfo temporalInfo) { + this.temporalInfo = temporalInfo; + } + + public String getAppId() { + return appId; + } + + /** + * Actual http request Uri, this does not contribute to the key behavior, + * it is used solely for interoperability between @AMSPropertyProvider.MetricsRequest + * and the Cache. + * @return Request Uri + */ + public String getSpec() { + return spec; + } + + /** + * Set spec string. + * @param spec Request Uri + */ + public void setSpec(String spec) { + this.spec = spec; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + TimelineAppMetricCacheKey that = (TimelineAppMetricCacheKey) o; + + if (appId != null ? !appId.equals(that.appId) : that.appId != null) + return false; + if (metricNames != null ? !metricNames.equals(that.metricNames) : that.metricNames != null) + return false; + + return true; + } + + @Override + public int hashCode() { + int result = metricNames != null ? metricNames.hashCode() : 0; + result = 31 * result + (appId != null ? appId.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "TimelineAppMetricCacheKey {" + + "metricNames = " + metricNames + + ", appId = '" + appId + '\'' + + ", temporalInfo = " + temporalInfo + + ", uriInfo = " + spec + + '}'; + } +}