/*
 * 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.kafka.streams.processor.internals.metrics;

import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Gauge;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.CumulativeCount;
import org.apache.kafka.common.metrics.stats.CumulativeSum;
import org.apache.kafka.common.metrics.stats.Max;
import org.apache.kafka.common.metrics.stats.Min;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.metrics.stats.Value;
import org.apache.kafka.common.metrics.stats.WindowedCount;
import org.apache.kafka.common.metrics.stats.WindowedSum;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecordingTrigger;

import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public class StreamsMetricsImpl implements StreamsMetrics {

    public enum Version {
        LATEST
    }

    static class ImmutableMetricValue<T> implements Gauge<T> {
        private final T value;

        public ImmutableMetricValue(final T value) {
            this.value = value;
        }

        @Override
        public T value(final MetricConfig config, final long now) {
            return value;
        }

        @Override
        public boolean equals(final Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            final ImmutableMetricValue<?> that = (ImmutableMetricValue<?>) o;
            return Objects.equals(value, that.value);
        }

        @Override
        public int hashCode() {
            return Objects.hash(value);
        }
    }

    private final Metrics metrics;
    private final Map<Sensor, Sensor> parentSensors;
    private final String clientId;

    private final Version version;
    private final Deque<MetricName> clientLevelMetrics = new LinkedList<>();
    private final Deque<String> clientLevelSensors = new LinkedList<>();
    private final Map<String, Deque<MetricName>> threadLevelMetrics = new HashMap<>();
    private final Map<String, Deque<String>> threadLevelSensors = new HashMap<>();
    private final Map<String, Deque<String>> taskLevelSensors = new HashMap<>();
    private final Map<String, Deque<String>> nodeLevelSensors = new HashMap<>();
    private final Map<String, Deque<String>> topicLevelSensors = new HashMap<>();
    private final Map<String, Deque<String>> cacheLevelSensors = new HashMap<>();
    private final ConcurrentMap<String, Deque<String>> storeLevelSensors = new ConcurrentHashMap<>();
    private final ConcurrentMap<String, Deque<MetricName>> storeLevelMetrics = new ConcurrentHashMap<>();

    private final RocksDBMetricsRecordingTrigger rocksDBMetricsRecordingTrigger;

    private static final String SENSOR_PREFIX_DELIMITER = ".";
    private static final String SENSOR_NAME_DELIMITER = ".s.";
    private static final String SENSOR_TASK_LABEL = "task";
    private static final String SENSOR_NODE_LABEL = "node";
    private static final String SENSOR_TOPIC_LABEL = "topic";
    private static final String SENSOR_CACHE_LABEL = "cache";
    private static final String SENSOR_STORE_LABEL = "store";
    private static final String SENSOR_ENTITY_LABEL = "entity";
    private static final String SENSOR_EXTERNAL_LABEL = "external";
    private static final String SENSOR_INTERNAL_LABEL = "internal";

    public static final String CLIENT_ID_TAG = "client-id";
    public static final String PROCESS_ID_TAG = "process-id";
    public static final String APPLICATION_ID_TAG = "application-id";
    public static final String THREAD_ID_TAG = "thread-id";
    public static final String TASK_ID_TAG = "task-id";
    public static final String PROCESSOR_NODE_ID_TAG = "processor-node-id";
    public static final String TOPIC_NAME_TAG = "topic";
    public static final String STORE_ID_TAG = "state-id";
    public static final String RECORD_CACHE_ID_TAG = "record-cache-id";

    public static final String ROLLUP_VALUE = "all";

    public static final String LATENCY_SUFFIX = "-latency";
    public static final String RECORDS_SUFFIX = "-records";
    public static final String AVG_SUFFIX = "-avg";
    public static final String MAX_SUFFIX = "-max";
    public static final String MIN_SUFFIX = "-min";
    public static final String RATE_SUFFIX = "-rate";
    public static final String TOTAL_SUFFIX = "-total";
    public static final String RATIO_SUFFIX = "-ratio";

    public static final String GROUP_PREFIX_WO_DELIMITER = "stream";
    public static final String GROUP_PREFIX = GROUP_PREFIX_WO_DELIMITER + "-";
    public static final String GROUP_SUFFIX = "-metrics";
    public static final String CLIENT_LEVEL_GROUP = GROUP_PREFIX_WO_DELIMITER + GROUP_SUFFIX;
    public static final String THREAD_LEVEL_GROUP = GROUP_PREFIX + "thread" + GROUP_SUFFIX;
    public static final String TASK_LEVEL_GROUP = GROUP_PREFIX + "task" + GROUP_SUFFIX;
    public static final String PROCESSOR_NODE_LEVEL_GROUP = GROUP_PREFIX + "processor-node" + GROUP_SUFFIX;
    public static final String TOPIC_LEVEL_GROUP = GROUP_PREFIX + "topic" + GROUP_SUFFIX;
    public static final String STATE_STORE_LEVEL_GROUP = GROUP_PREFIX + "state" + GROUP_SUFFIX;
    public static final String CACHE_LEVEL_GROUP = GROUP_PREFIX + "record-cache" + GROUP_SUFFIX;

    public static final String OPERATIONS = " operations";
    public static final String TOTAL_DESCRIPTION = "The total number of ";
    public static final String RATE_DESCRIPTION = "The average per-second number of ";
    public static final String RATIO_DESCRIPTION = "The fraction of time the thread spent on ";
    public static final String AVG_LATENCY_DESCRIPTION = "The average latency of ";
    public static final String MAX_LATENCY_DESCRIPTION = "The maximum latency of ";
    public static final String LATENCY_DESCRIPTION_SUFFIX = " in milliseconds";
    public static final String RATE_DESCRIPTION_PREFIX = "The average number of ";
    public static final String RATE_DESCRIPTION_SUFFIX = " per second";

    public static final String RECORD_E2E_LATENCY = "record-e2e-latency";
    public static final String RECORD_E2E_LATENCY_DESCRIPTION_SUFFIX =
        "end-to-end latency of a record, measuring by comparing the record timestamp with the "
            + "system time when it has been fully processed by the node";
    public static final String RECORD_E2E_LATENCY_AVG_DESCRIPTION = "The average " + RECORD_E2E_LATENCY_DESCRIPTION_SUFFIX;
    public static final String RECORD_E2E_LATENCY_MIN_DESCRIPTION = "The minimum " + RECORD_E2E_LATENCY_DESCRIPTION_SUFFIX;
    public static final String RECORD_E2E_LATENCY_MAX_DESCRIPTION = "The maximum " + RECORD_E2E_LATENCY_DESCRIPTION_SUFFIX;

    public StreamsMetricsImpl(final Metrics metrics,
                              final String clientId,
                              final Time time) {
        Objects.requireNonNull(metrics, "Metrics cannot be null");
        this.metrics = metrics;
        this.clientId = clientId;
        version = Version.LATEST;
        rocksDBMetricsRecordingTrigger = new RocksDBMetricsRecordingTrigger(time);

        this.parentSensors = new HashMap<>();
    }

    public Version version() {
        return version;
    }

    public Metrics metricsRegistry() {
        return metrics;
    }

    public RocksDBMetricsRecordingTrigger rocksDBMetricsRecordingTrigger() {
        return rocksDBMetricsRecordingTrigger;
    }

    public <T> void addClientLevelImmutableMetric(final String name,
                                                  final String description,
                                                  final RecordingLevel recordingLevel,
                                                  final T value) {
        addClientLevelImmutableMetric(name, description, Collections.emptyMap(), recordingLevel, value);
    }

    public <T> void addClientLevelImmutableMetric(final String name,
                                                  final String description,
                                                  final Map<String, String> additionalTags,
                                                  final RecordingLevel recordingLevel,
                                                  final T value) {
        final MetricName metricName = metrics.metricName(
            name,
            CLIENT_LEVEL_GROUP,
            description,
            clientLevelTagMap(additionalTags)
        );
        final MetricConfig metricConfig = new MetricConfig().recordLevel(recordingLevel);
        synchronized (clientLevelMetrics) {
            metrics.addMetric(metricName, metricConfig, new ImmutableMetricValue<>(value));
            clientLevelMetrics.push(metricName);
        }
    }

    public <T> void addClientLevelMutableMetric(final String name,
                                                final String description,
                                                final RecordingLevel recordingLevel,
                                                final Gauge<T> valueProvider) {
        addClientLevelMutableMetric(name, description, Collections.emptyMap(), recordingLevel, valueProvider);
    }

    public <T> void addClientLevelMutableMetric(final String name,
                                                final String description,
                                                final Map<String, String> additionalTags,
                                                final RecordingLevel recordingLevel,
                                                final Gauge<T> valueProvider) {
        final MetricName metricName = metrics.metricName(
            name,
            CLIENT_LEVEL_GROUP,
            description,
            clientLevelTagMap(additionalTags)
        );
        final MetricConfig metricConfig = new MetricConfig().recordLevel(recordingLevel);
        synchronized (clientLevelMetrics) {
            metrics.addMetric(metricName, metricConfig, valueProvider);
            clientLevelMetrics.push(metricName);
        }
    }

    public <T> void addThreadLevelImmutableMetric(final String name,
        final String description,
        final String threadId,
        final T value) {
        final MetricName metricName = metrics.metricName(
            name, THREAD_LEVEL_GROUP, description, threadLevelTagMap(threadId));
        synchronized (threadLevelMetrics) {
            threadLevelMetrics.computeIfAbsent(
                threadSensorPrefix(threadId),
                tid -> new LinkedList<>()
            ).add(metricName);
            metrics.addMetric(metricName, new ImmutableMetricValue<>(value));
        }
    }

    public <T> void addThreadLevelMutableMetric(final String name,
                                                final String description,
                                                final String threadId,
                                                final Gauge<T> valueProvider) {
        addThreadLevelMutableMetric(name, description, threadId, Collections.emptyMap(), valueProvider);
    }
    public <T> void addThreadLevelMutableMetric(final String name,
                                                final String description,
                                                final String threadId,
                                                final Map<String, String> additionalTags,
                                                final Gauge<T> valueProvider) {
        final MetricName metricName = metrics.metricName(
            name, THREAD_LEVEL_GROUP, description, threadLevelTagMap(threadId, additionalTags));
        synchronized (threadLevelMetrics) {
            threadLevelMetrics.computeIfAbsent(
                threadSensorPrefix(threadId),
                tid -> new LinkedList<>()
            ).add(metricName);
            metrics.addMetric(metricName, valueProvider);
        }
    }

    public final Sensor clientLevelSensor(final String sensorName,
                                          final RecordingLevel recordingLevel,
                                          final Sensor... parents) {
        synchronized (clientLevelSensors) {
            final String fullSensorName = CLIENT_LEVEL_GROUP + SENSOR_NAME_DELIMITER + sensorName;
            final Sensor sensor = metrics.getSensor(fullSensorName);
            if (sensor == null) {
                clientLevelSensors.push(fullSensorName);
                return metrics.sensor(fullSensorName, recordingLevel, parents);
            }
            return sensor;
        }
    }

    public final Sensor threadLevelSensor(final String threadId,
                                          final String sensorSuffix,
                                          final RecordingLevel recordingLevel,
                                          final Sensor... parents) {
        final String sensorPrefix = threadSensorPrefix(threadId);
        synchronized (threadLevelSensors) {
            return getSensors(threadLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    private String threadSensorPrefix(final String threadId) {
        return SENSOR_INTERNAL_LABEL + SENSOR_PREFIX_DELIMITER + threadId;
    }

    public Map<String, String> clientLevelTagMap() {
        return clientLevelTagMap(Collections.emptyMap());
    }

    public Map<String, String> clientLevelTagMap(final Map<String, String> additionalTags) {
        final Map<String, String> tagMap = new LinkedHashMap<>(additionalTags);
        tagMap.put(CLIENT_ID_TAG, clientId);
        return tagMap;
    }

    public Map<String, String> threadLevelTagMap(final String threadId) {
        return threadLevelTagMap(threadId, Collections.emptyMap());
    }

    public Map<String, String> threadLevelTagMap(final String threadId, final Map<String, String> additionalTags) {
        final Map<String, String> tagMap = new LinkedHashMap<>(additionalTags);
        tagMap.put(THREAD_ID_TAG, threadId);
        return tagMap;
    }

    public final void removeAllClientLevelSensorsAndMetrics() {
        removeAllClientLevelSensors();
        removeAllClientLevelMetrics();
    }

    private void removeAllClientLevelMetrics() {
        synchronized (clientLevelMetrics) {
            while (!clientLevelMetrics.isEmpty()) {
                metrics.removeMetric(clientLevelMetrics.pop());
            }
        }
    }

    private void removeAllClientLevelSensors() {
        synchronized (clientLevelSensors) {
            while (!clientLevelSensors.isEmpty()) {
                metrics.removeSensor(clientLevelSensors.pop());
            }
        }
    }

    public final void removeAllThreadLevelSensors(final String threadId) {
        final String key = threadSensorPrefix(threadId);
        synchronized (threadLevelSensors) {
            final Deque<String> sensors = threadLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    public final void removeAllThreadLevelMetrics(final String threadId) {
        synchronized (threadLevelMetrics) {
            final Deque<MetricName> names = threadLevelMetrics.remove(threadSensorPrefix(threadId));
            while (names != null && !names.isEmpty()) {
                metrics.removeMetric(names.pop());
            }
        }
    }

    public void removeMetric(final MetricName metricName) {
        metrics.removeMetric(metricName);
    }

    public void removeStoreLevelMetric(final MetricName metricName) {
        metrics.removeMetric(metricName);

        final List<String> metricsScopeCandidates = metricName.tags().keySet().stream()
            .filter(tag -> !tag.equals(THREAD_ID_TAG) && !tag.equals(TASK_ID_TAG))
            .collect(Collectors.toList());
        if (metricsScopeCandidates.size() != 1) {
            // should never happen
            throw new IllegalStateException("Expected exactly one metric scope tag, but found " + metricsScopeCandidates);
        }

        final Deque<MetricName> metricsForStore = storeLevelMetrics.get(
            storeSensorPrefix(
                metricName.tags().get(THREAD_ID_TAG),
                metricName.tags().get(TASK_ID_TAG),
                metricName.tags().get(metricsScopeCandidates.get(0))
            )
        );

        if (metricsForStore != null) {
            metricsForStore.remove(metricName);
        }
    }

    public Map<String, String> taskLevelTagMap(final String threadId, final String taskId) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);
        return tagMap;
    }

    public Map<String, String> nodeLevelTagMap(final String threadId,
                                               final String taskId,
                                               final String processorNodeName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);
        tagMap.put(PROCESSOR_NODE_ID_TAG, processorNodeName);
        return tagMap;
    }

    public Map<String, String> topicLevelTagMap(final String threadId,
                                                final String taskId,
                                                final String processorNodeName,
                                                final String topicName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);
        tagMap.put(PROCESSOR_NODE_ID_TAG, processorNodeName);
        tagMap.put(TOPIC_NAME_TAG, topicName);
        return tagMap;
    }

    public Map<String, String> storeLevelTagMap(final String taskId,
                                                final String storeType,
                                                final String storeName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, Thread.currentThread().getName());
        tagMap.put(TASK_ID_TAG, taskId);
        tagMap.put(storeType + "-" + STORE_ID_TAG, storeName);
        return tagMap;
    }

    public final Sensor taskLevelSensor(final String threadId,
                                        final String taskId,
                                        final String sensorSuffix,
                                        final RecordingLevel recordingLevel,
                                        final Sensor... parents) {
        final String sensorPrefix = taskSensorPrefix(threadId, taskId);
        synchronized (taskLevelSensors) {
            return getSensors(taskLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    public final void removeAllTaskLevelSensors(final String threadId, final String taskId) {
        final String key = taskSensorPrefix(threadId, taskId);
        synchronized (taskLevelSensors) {
            final Deque<String> sensors = taskLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    private String taskSensorPrefix(final String threadId, final String taskId) {
        return threadSensorPrefix(threadId) + SENSOR_PREFIX_DELIMITER + SENSOR_TASK_LABEL + SENSOR_PREFIX_DELIMITER +
            taskId;
    }

    public Sensor nodeLevelSensor(final String threadId,
                                  final String taskId,
                                  final String processorNodeName,
                                  final String sensorSuffix,
                                  final Sensor.RecordingLevel recordingLevel,
                                  final Sensor... parents) {
        final String sensorPrefix = nodeSensorPrefix(threadId, taskId, processorNodeName);
        synchronized (nodeLevelSensors) {
            return getSensors(nodeLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    public final void removeAllNodeLevelSensors(final String threadId,
                                                final String taskId,
                                                final String processorNodeName) {
        final String key = nodeSensorPrefix(threadId, taskId, processorNodeName);
        synchronized (nodeLevelSensors) {
            final Deque<String> sensors = nodeLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    private String nodeSensorPrefix(final String threadId, final String taskId, final String processorNodeName) {
        return taskSensorPrefix(threadId, taskId)
            + SENSOR_PREFIX_DELIMITER + SENSOR_NODE_LABEL + SENSOR_PREFIX_DELIMITER + processorNodeName;
    }

    public Sensor topicLevelSensor(final String threadId,
                                   final String taskId,
                                   final String processorNodeName,
                                   final String topicName,
                                   final String sensorSuffix,
                                   final Sensor.RecordingLevel recordingLevel,
                                   final Sensor... parents) {
        final String sensorPrefix = topicSensorPrefix(threadId, taskId, processorNodeName, topicName);
        synchronized (topicLevelSensors) {
            return getSensors(topicLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
        }
    }

    public final void removeAllTopicLevelSensors(final String threadId,
                                                 final String taskId,
                                                 final String processorNodeName,
                                                 final String topicName) {
        final String key = topicSensorPrefix(threadId, taskId, processorNodeName, topicName);
        synchronized (topicLevelSensors) {
            final Deque<String> sensors = topicLevelSensors.remove(key);
            while (sensors != null && !sensors.isEmpty()) {
                metrics.removeSensor(sensors.pop());
            }
        }
    }

    private String topicSensorPrefix(final String threadId,
                                     final String taskId,
                                     final String processorNodeName,
                                     final String topicName) {
        return nodeSensorPrefix(threadId, taskId, processorNodeName)
            + SENSOR_PREFIX_DELIMITER + SENSOR_TOPIC_LABEL + SENSOR_PREFIX_DELIMITER + topicName;
    }

    public Sensor cacheLevelSensor(final String threadId,
                                   final String taskName,
                                   final String storeName,
                                   final String ratioName,
                                   final Sensor.RecordingLevel recordingLevel,
                                   final Sensor... parents) {
        // use ratio name as sensor suffix
        final String sensorPrefix = cacheSensorPrefix(threadId, taskName, storeName);
        synchronized (cacheLevelSensors) {
            return getSensors(cacheLevelSensors, ratioName, sensorPrefix, recordingLevel, parents);
        }
    }

    public Map<String, String> cacheLevelTagMap(final String threadId,
                                                final String taskId,
                                                final String storeName) {
        final Map<String, String> tagMap = new LinkedHashMap<>();
        tagMap.put(THREAD_ID_TAG, threadId);
        tagMap.put(TASK_ID_TAG, taskId);
        tagMap.put(RECORD_CACHE_ID_TAG, storeName);
        return tagMap;
    }

    public final void removeAllCacheLevelSensors(final String threadId, final String taskId, final String cacheName) {
        final String key = cacheSensorPrefix(threadId, taskId, cacheName);
        synchronized (cacheLevelSensors) {
            final Deque<String> strings = cacheLevelSensors.remove(key);
            while (strings != null && !strings.isEmpty()) {
                metrics.removeSensor(strings.pop());
            }
        }
    }

    private String cacheSensorPrefix(final String threadId, final String taskId, final String cacheName) {
        return taskSensorPrefix(threadId, taskId)
            + SENSOR_PREFIX_DELIMITER + SENSOR_CACHE_LABEL + SENSOR_PREFIX_DELIMITER + cacheName;
    }

    public final Sensor storeLevelSensor(final String taskId,
                                         final String storeName,
                                         final String sensorSuffix,
                                         final RecordingLevel recordingLevel,
                                         final Sensor... parents) {
        final String sensorPrefix = storeSensorPrefix(Thread.currentThread().getName(), taskId, storeName);
            // since the keys in the map storeLevelSensors contain the name of the current thread and threads only
            // access keys in which their name is contained, the value in the maps do not need to be thread safe
            // and we can use a LinkedList here.
            // TODO: In future, we could use thread local maps since each thread will exclusively access the set of keys
            //  that contain its name. Similar is true for the other metric levels. Thread-level metrics need some
            //  special attention, since they are created before the thread is constructed. The creation of those
            //  metrics could be moved into the run() method of the thread.
        return getSensors(storeLevelSensors, sensorSuffix, sensorPrefix, recordingLevel, parents);
    }

    public <T> MetricName addStoreLevelMutableMetric(final String taskId,
                                                     final String metricsScope,
                                                     final String storeName,
                                                     final String name,
                                                     final String description,
                                                     final RecordingLevel recordingLevel,
                                                     final Gauge<T> valueProvider) {
        final MetricName metricName = metrics.metricName(
            name,
            STATE_STORE_LEVEL_GROUP,
            description,
            storeLevelTagMap(taskId, metricsScope, storeName)
        );
        if (metrics.metric(metricName) == null) {
            metrics.addMetricIfAbsent(metricName, new MetricConfig().recordLevel(recordingLevel), valueProvider);
            final String key = storeSensorPrefix(Thread.currentThread().getName(), taskId, storeName);
            storeLevelMetrics.computeIfAbsent(key, ignored -> new LinkedList<>()).push(metricName);
        }

        return metricName;
    }

    public final void removeAllStoreLevelSensorsAndMetrics(final String taskId,
                                                           final String storeName) {
        final String threadId = Thread.currentThread().getName();
        removeAllStoreLevelSensors(threadId, taskId, storeName);
        removeAllStoreLevelMetrics(threadId, taskId, storeName);
    }

    private void removeAllStoreLevelSensors(final String threadId,
                                            final String taskId,
                                            final String storeName) {
        final String key = storeSensorPrefix(threadId, taskId, storeName);
        final Deque<String> sensors = storeLevelSensors.remove(key);
        while (sensors != null && !sensors.isEmpty()) {
            metrics.removeSensor(sensors.pop());
        }
    }

    private void removeAllStoreLevelMetrics(final String threadId,
                                            final String taskId,
                                            final String storeName) {
        final String key = storeSensorPrefix(threadId, taskId, storeName);
        final Deque<MetricName> metricNames = storeLevelMetrics.remove(key);
        while (metricNames != null && !metricNames.isEmpty()) {
            metrics.removeMetric(metricNames.pop());
        }
    }

    private String storeSensorPrefix(final String threadId,
                                     final String taskId,
                                     final String storeName) {
        return taskSensorPrefix(threadId, taskId)
            + SENSOR_PREFIX_DELIMITER + SENSOR_STORE_LABEL + SENSOR_PREFIX_DELIMITER + storeName;
    }

    @Override
    public Sensor addSensor(final String name, final Sensor.RecordingLevel recordingLevel) {
        return metrics.sensor(name, recordingLevel);
    }

    @Override
    public Sensor addSensor(final String name, final Sensor.RecordingLevel recordingLevel, final Sensor... parents) {
        return metrics.sensor(name, recordingLevel, parents);
    }

    @Override
    public Map<MetricName, ? extends Metric> metrics() {
        return Collections.unmodifiableMap(this.metrics.metrics());
    }

    private Map<String, String> customizedTags(final String threadId,
                                               final String scopeName,
                                               final String entityName,
                                               final String... tags) {
        final Map<String, String> tagMap = threadLevelTagMap(threadId);
        tagMap.put(scopeName + "-id", entityName);
        if (tags != null) {
            if ((tags.length % 2) != 0) {
                throw new IllegalArgumentException("Tags needs to be specified in key-value pairs");
            }
            for (int i = 0; i < tags.length; i += 2) {
                tagMap.put(tags[i], tags[i + 1]);
            }
        }
        return tagMap;
    }

    private Sensor customInvocationRateAndCountSensor(final String threadId,
                                                      final String groupName,
                                                      final String entityName,
                                                      final String operationName,
                                                      final Map<String, String> tags,
                                                      final Sensor.RecordingLevel recordingLevel) {
        final Sensor sensor = metrics.sensor(externalChildSensorName(threadId, operationName, entityName), recordingLevel);
        addInvocationRateAndCountToSensor(
            sensor,
            groupName,
            tags,
            operationName,
            RATE_DESCRIPTION_PREFIX + operationName + OPERATIONS + RATE_DESCRIPTION_SUFFIX,
            TOTAL_DESCRIPTION + operationName + OPERATIONS
        );
        return sensor;
    }

    @Override
    public Sensor addLatencyRateTotalSensor(final String scopeName,
                                            final String entityName,
                                            final String operationName,
                                            final Sensor.RecordingLevel recordingLevel,
                                            final String... tags) {
        final String threadId = Thread.currentThread().getName();
        final String group = groupNameFromScope(scopeName);
        final Map<String, String> tagMap = customizedTags(threadId, scopeName, entityName, tags);
        final Sensor sensor =
            customInvocationRateAndCountSensor(threadId, group, entityName, operationName, tagMap, recordingLevel);
        addAvgAndMaxToSensor(
            sensor,
            group,
            tagMap,
            operationName + LATENCY_SUFFIX,
            AVG_LATENCY_DESCRIPTION + operationName,
            MAX_LATENCY_DESCRIPTION + operationName
        );

        return sensor;
    }

    @Override
    public Sensor addRateTotalSensor(final String scopeName,
                                     final String entityName,
                                     final String operationName,
                                     final Sensor.RecordingLevel recordingLevel,
                                     final String... tags) {
        final String threadId = Thread.currentThread().getName();
        final Map<String, String> tagMap = customizedTags(threadId, scopeName, entityName, tags);
        return customInvocationRateAndCountSensor(
            threadId,
            groupNameFromScope(scopeName),
            entityName,
            operationName,
            tagMap,
            recordingLevel
        );
    }

    private String externalChildSensorName(final String threadId, final String operationName, final String entityName) {
        return SENSOR_EXTERNAL_LABEL + SENSOR_PREFIX_DELIMITER + threadId
            + SENSOR_PREFIX_DELIMITER + SENSOR_ENTITY_LABEL + SENSOR_PREFIX_DELIMITER + entityName
            + SENSOR_NAME_DELIMITER + operationName;
    }

    public static void addAvgAndMaxToSensor(final Sensor sensor,
                                            final String group,
                                            final Map<String, String> tags,
                                            final String gaugeName,
                                            final String descriptionOfAvg,
                                            final String descriptionOfMax) {
        sensor.add(
            new MetricName(
                gaugeName + AVG_SUFFIX,
                group,
                descriptionOfAvg,
                tags),
            new Avg()
        );
        sensor.add(
            new MetricName(
                gaugeName + MAX_SUFFIX,
                group,
                descriptionOfMax,
                tags),
            new Max()
        );
    }

    public static void addMinAndMaxToSensor(final Sensor sensor,
                                            final String group,
                                            final Map<String, String> tags,
                                            final String operation,
                                            final String descriptionOfMin,
                                            final String descriptionOfMax) {
        sensor.add(
            new MetricName(
                operation + MIN_SUFFIX,
                group,
                descriptionOfMin,
                tags),
            new Min()
        );

        sensor.add(
            new MetricName(
                operation + MAX_SUFFIX,
                group,
                descriptionOfMax,
                tags),
            new Max()
        );
    }

    public static void addAvgAndMaxLatencyToSensor(final Sensor sensor,
                                                   final String group,
                                                   final Map<String, String> tags,
                                                   final String operation) {
        sensor.add(
            new MetricName(
                operation + "-latency-avg",
                group,
                AVG_LATENCY_DESCRIPTION + operation + " operation.",
                tags),
            new Avg()
        );
        sensor.add(
            new MetricName(
                operation + "-latency-max",
                group,
                MAX_LATENCY_DESCRIPTION + operation + " operation.",
                tags),
            new Max()
        );
    }

    public static void addAvgAndMinAndMaxToSensor(final Sensor sensor,
                                                  final String group,
                                                  final Map<String, String> tags,
                                                  final String gaugeName,
                                                  final String descriptionOfAvg,
                                                  final String descriptionOfMin,
                                                  final String descriptionOfMax) {
        addAvgAndMaxToSensor(sensor, group, tags, gaugeName, descriptionOfAvg, descriptionOfMax);
        sensor.add(
            new MetricName(
                gaugeName + MIN_SUFFIX,
                group,
                descriptionOfMin,
                tags),
            new Min()
        );
    }

    public static void addInvocationRateAndCountToSensor(final Sensor sensor,
                                                         final String group,
                                                         final Map<String, String> tags,
                                                         final String operation,
                                                         final String descriptionOfRate,
                                                         final String descriptionOfCount) {
        addInvocationRateToSensor(sensor, group, tags, operation, descriptionOfRate);
        sensor.add(
            new MetricName(
                operation + TOTAL_SUFFIX,
                group,
                descriptionOfCount,
                tags
            ),
            new CumulativeCount()
        );
    }

    public static void addInvocationRateToSensor(final Sensor sensor,
                                                 final String group,
                                                 final Map<String, String> tags,
                                                 final String operation,
                                                 final String descriptionOfRate) {
        sensor.add(
            new MetricName(
                operation + RATE_SUFFIX,
                group,
                descriptionOfRate,
                tags
            ),
            new Rate(TimeUnit.SECONDS, new WindowedCount())
        );
    }

    public static void addRateOfSumAndSumMetricsToSensor(final Sensor sensor,
                                                         final String group,
                                                         final Map<String, String> tags,
                                                         final String operation,
                                                         final String descriptionOfRate,
                                                         final String descriptionOfTotal) {
        addRateOfSumMetricToSensor(sensor, group, tags, operation, descriptionOfRate);
        addSumMetricToSensor(sensor, group, tags, operation, descriptionOfTotal);
    }

    public static void addRateOfSumMetricToSensor(final Sensor sensor,
                                                  final String group,
                                                  final Map<String, String> tags,
                                                  final String operation,
                                                  final String description) {
        sensor.add(new MetricName(operation + RATE_SUFFIX, group, description, tags),
                   new Rate(TimeUnit.SECONDS, new WindowedSum()));
    }

    public static void addSumMetricToSensor(final Sensor sensor,
                                            final String group,
                                            final Map<String, String> tags,
                                            final String operation,
                                            final String description) {
        addSumMetricToSensor(sensor, group, tags, operation, true, description);
    }

    public static void addSumMetricToSensor(final Sensor sensor,
                                            final String group,
                                            final Map<String, String> tags,
                                            final String operation,
                                            final boolean withSuffix,
                                            final String description) {
        sensor.add(
            new MetricName(
                withSuffix ? operation + TOTAL_SUFFIX : operation,
                group,
                description,
                tags
            ),
            new CumulativeSum()
        );
    }

    public static void addValueMetricToSensor(final Sensor sensor,
                                              final String group,
                                              final Map<String, String> tags,
                                              final String name,
                                              final String description) {
        sensor.add(new MetricName(name, group, description, tags), new Value());
    }

    public static void addAvgAndSumMetricsToSensor(final Sensor sensor,
                                                   final String group,
                                                   final Map<String, String> tags,
                                                   final String metricNamePrefix,
                                                   final String descriptionOfAvg,
                                                   final String descriptionOfTotal) {
        sensor.add(new MetricName(metricNamePrefix + AVG_SUFFIX, group, descriptionOfAvg, tags), new Avg());
        sensor.add(
            new MetricName(metricNamePrefix + TOTAL_SUFFIX, group, descriptionOfTotal, tags),
            new CumulativeSum()
        );
    }

    public static void addTotalCountAndSumMetricsToSensor(final Sensor sensor,
                                                          final String group,
                                                          final Map<String, String> tags,
                                                          final String countMetricNamePrefix,
                                                          final String sumMetricNamePrefix,
                                                          final String descriptionOfCount,
                                                          final String descriptionOfTotal) {
        sensor.add(
            new MetricName(countMetricNamePrefix + TOTAL_SUFFIX, group, descriptionOfCount, tags),
            new CumulativeCount()
        );
        sensor.add(
            new MetricName(sumMetricNamePrefix + TOTAL_SUFFIX, group, descriptionOfTotal, tags),
            new CumulativeSum()
        );
    }

    public static void maybeRecordSensor(final double value,
                                         final Time time,
                                         final Sensor sensor) {
        if (sensor.shouldRecord() && sensor.hasMetrics()) {
            sensor.record(value, time.milliseconds());
        }
    }

    public static void maybeMeasureLatency(final Runnable actionToMeasure,
                                           final Time time,
                                           final Sensor sensor) {
        if (sensor.shouldRecord() && sensor.hasMetrics()) {
            final long startNs = time.nanoseconds();
            try {
                actionToMeasure.run();
            } finally {
                sensor.record(time.nanoseconds() - startNs);
            }
        } else {
            actionToMeasure.run();
        }
    }

    public static <T> T maybeMeasureLatency(final Supplier<T> actionToMeasure,
                                            final Time time,
                                            final Sensor sensor) {
        if (sensor.shouldRecord() && sensor.hasMetrics()) {
            final long startNs = time.nanoseconds();
            try {
                return actionToMeasure.get();
            } finally {
                sensor.record(time.nanoseconds() - startNs);
            }
        } else {
            return actionToMeasure.get();
        }
    }

    private Sensor getSensors(final Map<String, Deque<String>> sensors,
                              final String sensorSuffix,
                              final String sensorPrefix,
                              final RecordingLevel recordingLevel,
                              final Sensor... parents) {
        final String fullSensorName = sensorPrefix + SENSOR_NAME_DELIMITER + sensorSuffix;
        final Sensor sensor = metrics.getSensor(fullSensorName);
        if (sensor == null) {
            sensors.computeIfAbsent(sensorPrefix, ignored -> new LinkedList<>()).push(fullSensorName);
            return metrics.sensor(fullSensorName, recordingLevel, parents);
        }
        return sensor;
    }

    /**
     * Deletes a sensor and its parents, if any
     */
    @Override
    public void removeSensor(final Sensor sensor) {
        Objects.requireNonNull(sensor, "Sensor is null");
        metrics.removeSensor(sensor.name());

        final Sensor parent = parentSensors.remove(sensor);
        if (parent != null) {
            metrics.removeSensor(parent.name());
        }
    }

    /**
     * Visible for testing
     */
    Map<Sensor, Sensor> parentSensors() {
        return Collections.unmodifiableMap(parentSensors);
    }

    private static String groupNameFromScope(final String scopeName) {
        return "stream-" + scopeName + "-metrics";
    }
}
