container/src/main/java/com/streamsets/datacollector/execution/alerts/MetricRuleEvaluatorHelper.java
public static Object getMetricValue(
    MetricRegistry metrics,
    String metricId,
    MetricType metricType,
    MetricElement metricElement
  ) throws ObserverException {
    // We moved the logic of CURRENT_BATCH_AGE and TIME_IN_CURRENT_STAGE due to multi-threaded framework
    if(metricElement.isOneOf(MetricElement.CURRENT_BATCH_AGE, MetricElement.TIME_IN_CURRENT_STAGE)) {
      switch (metricElement) {
        case CURRENT_BATCH_AGE:
          return getTimeFromRunner(metrics, PipeRunner.METRIC_BATCH_START_TIME);
        case TIME_IN_CURRENT_STAGE:
          return getTimeFromRunner(metrics, PipeRunner.METRIC_STAGE_START_TIME);
        default:
          throw new IllegalStateException(Utils.format("Unknown metric type '{}'", metricType));
      }
    }

    // Default path
    Metric metric = getMetric(
        metrics,
        metricId,
        metricType
    );

    if(metric != null) {
      return getMetricValue(metricElement, metricType, metric);
    }

    return null;
  }
public static Object getMetricValue(
    MetricRegistry metrics,
    String metricId,
    MetricType metricType,
    MetricElement metricElement
  ) throws ObserverException {
    // We moved the logic of CURRENT_BATCH_AGE and TIME_IN_CURRENT_STAGE due to multi-threaded framework
    if(metricElement.isOneOf(MetricElement.CURRENT_BATCH_AGE, MetricElement.TIME_IN_CURRENT_STAGE)) {
      switch (metricElement) {
        case CURRENT_BATCH_AGE:
          return getTimeFromRunner(metrics, PipeRunner.METRIC_BATCH_START_TIME);
        case TIME_IN_CURRENT_STAGE:
          return getTimeFromRunner(metrics, PipeRunner.METRIC_STAGE_START_TIME);
        default:
          throw new IllegalStateException(Utils.format("Unknown metric type '{}'", metricType));
      }
    }

    // Default path
    Metric metric = getMetric(
        metrics,
        metricId,
        metricType
    );

    if(metric != null) {
      return getMetricValue(metricElement, metricType, metric);
    }

    return null;
  }
public static Object getMetricValue ( MetricRegistry metrics , String metricId , MetricType metricType , MetricElement metricElement ) throws ObserverException { // We moved the logic of CURRENT_BATCH_AGE and TIME_IN_CURRENT_STAGE due to multi-threaded framework if ( metricElement . isOneOf ( MetricElement . CURRENT_BATCH_AGE , MetricElement . TIME_IN_CURRENT_STAGE ) ) { switch ( metricElement ) { case CURRENT_BATCH_AGE : return getTimeFromRunner ( metrics , PipeRunner . METRIC_BATCH_START_TIME ) ; case TIME_IN_CURRENT_STAGE : return getTimeFromRunner ( metrics , PipeRunner . METRIC_STAGE_START_TIME ) ; default : throw new IllegalStateException ( Utils . format ( "Unknown metric type '{}'" , metricType ) ) ; } } // Default path Metric metric = getMetric ( metrics , metricId , metricType ) ; if ( metric != null ) { return getMetricValue ( metricElement , metricType , metric ) ; } return null ; }
Get metric value for given rule evaluation.

@param metrics Metric Registry for the pipeline.
@param metricId Name of the metric where the value is "usually" stored. This method cover mappings of metrics
that got historically moved.
@param metricType Type of the metric
@param metricElement Value that caller needs in order to assert the right condition.
@return Requested metric value or null if it doesn't exists
@throws ObserverException
Get metric value for given rule evaluation .