@Override
  public <T> T timeControllerExecution(ControllerExecution<T> execution) throws Exception {
    log.debug("Time controller execution");
    final String name = execution.controllerName();
    final ResourceID resourceID = execution.resourceID();
    final Optional<String> namespaceOptional = resourceID.getNamespace();
    final Map<String, Object> metadata = execution.metadata();
    final Optional<Class<? extends BaseResource<?, ?, ?, ?, ?>>> resourceClass =
        getResourceClass(metadata);
    final String execName = execution.name();

    long startTime = clock.getTimeMillis();
    try {
      T result = execution.execute();
      final String successType = execution.successTypeName(result);
      if (resourceClass.isPresent()) {
        getHistogram(resourceClass.get(), name, execName, successType).update(toSeconds(startTime));
        getCounter(resourceClass.get(), name, execName, SUCCESS, successType).inc();
        if (namespaceOptional.isPresent()) {
          getHistogram(resourceClass.get(), namespaceOptional.get(), name, execName, successType)
              .update(toSeconds(startTime));
          getCounter(
                  resourceClass.get(),
                  namespaceOptional.get(),
                  name,
                  execName,
                  SUCCESS,
                  successType)
              .inc();
        }
      }
      return result;
    } catch (Exception e) {
      log.error(
          "Controller execution failed for resource {}, metadata {}", resourceID, metadata, e);
      final String exception = e.getClass().getSimpleName();
      if (resourceClass.isPresent()) {
        getHistogram(resourceClass.get(), name, execName, FAILURE).update(toSeconds(startTime));
        getCounter(resourceClass.get(), name, execName, FAILURE, EXCEPTION, exception).inc();
        if (namespaceOptional.isPresent()) {
          getHistogram(resourceClass.get(), namespaceOptional.get(), name, execName, FAILURE)
              .update(toSeconds(startTime));
          getCounter(
                  resourceClass.get(),
                  namespaceOptional.get(),
                  name,
                  execName,
                  FAILURE,
                  EXCEPTION,
                  exception)
              .inc();
        }
      }
      throw e;
    }
  }
@Test
  void testTimeControllerExecution() throws Exception {
    TestingExecutionBase<SparkApplication> successExecution = new TestingExecutionBase<>();
    operatorMetrics.timeControllerExecution(successExecution);
    Map<String, Metric> metrics = operatorMetrics.metricRegistry().getMetrics();
    assertEquals(4, metrics.size());
    assertTrue(metrics.containsKey("sparkapplication.test-controller.reconcile.both"));
    assertTrue(metrics.containsKey("sparkapplication.testns.test-controller.reconcile.both"));
    assertTrue(metrics.containsKey("sparkapplication.test-controller.reconcile.success.both"));
    assertTrue(
        metrics.containsKey("sparkapplication.testns.test-controller.reconcile.success.both"));

    FooTestingExecutionBase<SparkApplication> failedExecution = new FooTestingExecutionBase<>();
    try {
      operatorMetrics.timeControllerExecution(failedExecution);
    } catch (Exception e) {
      assertEquals(e.getMessage(), "Foo exception");
      assertEquals(8, metrics.size());
      assertTrue(metrics.containsKey("sparkapplication.test-controller.reconcile.failure"));
      assertTrue(
          metrics.containsKey(
              "sparkapplication.test-controller.reconcile.failure.exception"
                  + ".nosuchfieldexception"));
      assertTrue(metrics.containsKey("sparkapplication.testns.test-controller.reconcile.failure"));
      assertTrue(
          metrics.containsKey(
              "sparkapplication.testns.test-controller.reconcile.failure."
                  + "exception.nosuchfieldexception"));
    }
  }