/*
 * 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.nifi.controller.status.history;

import org.apache.nifi.controller.status.ProcessingPerformanceStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

public enum ProcessorStatusDescriptor {
    BYTES_READ(
            "bytesRead",
            "Bytes Read (5 mins)",
            "The total number of bytes read from the Content Repository by this Processor in the past 5 minutes",
            Formatter.DATA_SIZE,
            ProcessorStatus::getBytesRead),

    BYTES_WRITTEN(
            "bytesWritten",
            "Bytes Written (5 mins)",
            "The total number of bytes written to the Content Repository by this Processor in the past 5 minutes",
            Formatter.DATA_SIZE,
            ProcessorStatus::getBytesWritten),

    BYTES_TRANSFERRED(
            "bytesTransferred",
            "Bytes Transferred (5 mins)",
            "The total number of bytes read from or written to the Content Repository by this Processor in the past 5 minutes",
            Formatter.DATA_SIZE,
                s -> s.getBytesRead() + s.getBytesWritten()),

    INPUT_BYTES(
            "inputBytes",
            "Bytes In (5 mins)",
            "The cumulative size of all FlowFiles that this Processor has pulled from its queues in the past 5 minutes",
            Formatter.DATA_SIZE,
            ProcessorStatus::getInputBytes),

    INPUT_COUNT(
            "inputCount",
            "FlowFiles In (5 mins)",
            "The number of FlowFiles that this Processor has pulled from its queues in the past 5 minutes",
            Formatter.COUNT,
                s -> (long) s.getInputCount()),

    OUTPUT_BYTES(
            "outputBytes",
            "Bytes Out (5 mins)",
            "The cumulative size of all FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes",
            Formatter.DATA_SIZE,
            ProcessorStatus::getOutputBytes),

    OUTPUT_COUNT(
            "outputCount",
            "FlowFiles Out (5 mins)",
            "The number of FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes",
            Formatter.COUNT,
                s -> (long) s.getOutputCount()),

    TASK_COUNT(
            "taskCount",
            "Tasks (5 mins)",
            "The number of tasks that this Processor has completed in the past 5 minutes",
            Formatter.COUNT,
                s -> (long) s.getInvocations()),

    TASK_MILLIS(
            "taskMillis",
            "Total Task Duration (5 mins)",
            "The total number of thread-milliseconds that the Processor has used to complete its tasks in the past 5 minutes",
            Formatter.DURATION,
                s -> TimeUnit.MILLISECONDS.convert(s.getProcessingNanos(), TimeUnit.NANOSECONDS)),

    TASK_NANOS(
            "taskNanos",
            "Total Task Time (nanos)",
            "The total number of thread-nanoseconds that the Processor has used to complete its tasks in the past 5 minutes",
            Formatter.COUNT,
            ProcessorStatus::getProcessingNanos,
            false),

    FLOWFILES_REMOVED(
            "flowFilesRemoved",
            "FlowFiles Removed (5 mins)",
            "The total number of FlowFiles removed by this Processor in the last 5 minutes",
            Formatter.COUNT,
                s -> (long) s.getFlowFilesRemoved()),

    AVERAGE_LINEAGE_DURATION(
            "averageLineageDuration",
            "Average Lineage Duration (5 mins)",
            "The average amount of time that a FlowFile took to process (from receipt until this Processor finished processing it) in the past 5 minutes.",
            Formatter.DURATION,
                s -> s.getAverageLineageDuration(TimeUnit.MILLISECONDS),
            new ValueReducer<>() {
        @Override
        public Long reduce(final List<StatusSnapshot> values) {
            long millis = 0L;
            long count = 0;

            for (final StatusSnapshot snapshot : values) {
                final long removed = snapshot.getStatusMetric(FLOWFILES_REMOVED.getDescriptor());
                final long outputCount = snapshot.getStatusMetric(OUTPUT_COUNT.getDescriptor());
                final long processed = removed + outputCount;

                count += processed;

                final long avgMillis = snapshot.getStatusMetric(AVERAGE_LINEAGE_DURATION.getDescriptor());
                final long totalMillis = avgMillis * processed;
                millis += totalMillis;
            }

            return count == 0 ? 0 : millis / count;
        }
    },
            true
    ),

    AVERAGE_TASK_NANOS(
            "averageTaskNanos",
            "Average Task Duration (nanoseconds)",
            "The average number of nanoseconds it took this Processor to complete a task, over the past 5 minutes",
            Formatter.COUNT,
                s -> s.getInvocations() == 0 ? 0 : s.getProcessingNanos() / s.getInvocations(),
            new ValueReducer<>() {
        @Override
        public Long reduce(final List<StatusSnapshot> values) {
            long procNanos = 0L;
            int invocations = 0;

            for (final StatusSnapshot snapshot : values) {
                final Long taskNanos = snapshot.getStatusMetric(TASK_NANOS.getDescriptor());
                if (taskNanos != null) {
                    procNanos += taskNanos;
                }

                final Long taskInvocations = snapshot.getStatusMetric(TASK_COUNT.getDescriptor());
                if (taskInvocations != null) {
                    invocations += taskInvocations.intValue();
                }
            }

            if (invocations == 0) {
                return 0L;
            }

            return procNanos / invocations;
        }
    },
            true
    ),

    CPU_MILLIS(
            "cpuTime",
            "CPU Time (5 mins)",
            "The total amount of time that the Processor has used the CPU in the past 5 minutes. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.DURATION,
                status -> nanosToMillis(status, ProcessingPerformanceStatus::getCpuDuration)
    ),

    CPU_PERCENTAGE(
            "cpuPercentage",
            "CPU Percentage (5 mins)",
            "Of the time that the Processor was running in the past 5 minutes, the percentage of that time that the Processor was using the CPU. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.COUNT,
                status -> nanosValue(status, ProcessingPerformanceStatus::getCpuDuration) * 100 / Math.max(1, status.getProcessingNanos()),
            processingPercentage(CPU_MILLIS.getDescriptor()),
            true
    ),

    CONTENT_REPO_READ_MILLIS(
            "contentRepoReadTime",
            "Content Repo Read Time (5 mins)",
            "The amount of time that the Processor has spent reading from the Content Repository in the past 5 minutes. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.DURATION,
                status -> nanosToMillis(status, ProcessingPerformanceStatus::getContentReadDuration)
    ),

    CONTENT_REPO_READ_PERCENTAGE(
            "contentRepoReadPercentage",
            "Content Repo Read Percentage (5 mins)",
            "Of the time that the Processor was running in the past 5 minutes, the percentage of that time that the Processor was reading from the Content Repository. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.COUNT,
                status -> nanosValue(status, ProcessingPerformanceStatus::getContentReadDuration) * 100 / Math.max(1, status.getProcessingNanos()),
            processingPercentage(CONTENT_REPO_READ_MILLIS.getDescriptor()),
            true
    ),

    CONTENT_REPO_WRITE_MILLIS(
            "contentRepoWriteTime",
            "Content Repo Write Time (5 mins)",
            "The total amount of time that the Processor has spent writing to the Content Repository in the past 5 minutes. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.DURATION,
                status -> nanosToMillis(status, ProcessingPerformanceStatus::getContentWriteDuration)
    ),

    CONTENT_REPO_WRITE_PERCENTAGE(
            "contentRepoWritePercentage",
            "Content Repo Write Percentage (5 mins)",
            "Of the time that the Processor was running in the past 5 minutes, the percentage of that time that the Processor was writing to the Content Repository. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.COUNT,
                status -> nanosValue(status, ProcessingPerformanceStatus::getContentWriteDuration) * 100 / Math.max(1, status.getProcessingNanos()),
            processingPercentage(CONTENT_REPO_WRITE_MILLIS.getDescriptor()),
            true
    ),

    SESSION_COMMIT_MILLIS(
            "sessionCommitTime",
            "Session Commit Time (5 mins)",
            "The total amount of time that the Processor has spent waiting for the framework to commit its ProcessSession in the past 5 minutes. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.DURATION,
                status -> nanosToMillis(status, ProcessingPerformanceStatus::getSessionCommitDuration)
    ),

    SESSION_COMMIT_PERCENTAGE(
            "sessionCommitPercentage",
            "Session Commit Percentage (5 mins)",
            "Of the time that the Processor was running in the past 5 minutes, the percentage of that time that the Processor was waiting for hte framework to commit its ProcessSession. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.COUNT,
                status -> nanosValue(status, ProcessingPerformanceStatus::getSessionCommitDuration) * 100 / Math.max(1, status.getProcessingNanos()),
            processingPercentage(SESSION_COMMIT_MILLIS.getDescriptor()),
            true
    ),

    GARBAGE_COLLECTION_MILLIS(
            "garbageCollectionTime",
            "Garbage Collection Time (5 mins)",
            "The total amount of time that the Processor has spent blocked on Garbage Collection in the past 5 minutes. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.DURATION,
                status -> {
                    final ProcessingPerformanceStatus perfStatus = status.getProcessingPerformanceStatus();
                    if (perfStatus == null) {
                        return 0L;
                    }
                    // Garbage Collection is reported in milliseconds rather than nanos.
                    return perfStatus.getGarbageCollectionDuration();
                }
    ),

    GARBAGE_COLLECTION_PERCENTAGE(
            "garbageCollectionPercentage",
            "Garbage Collection Percentage (5 mins)",
            "Of the time that the Processor was running in the past 5 minutes, the percentage of that time that the Processor spent blocked on Garbage Collection. " +
            "Note, this metric may be unavailable, depending on configuration of the `nifi.performance.tracking.percentage` property.",
            Formatter.COUNT,
                status -> {
                    final ProcessingPerformanceStatus perfStatus = status.getProcessingPerformanceStatus();
                    if (perfStatus == null) {
                        return 0L;
                    }
                    // Garbage Collection is reported in milliseconds rather than nanos.
                    final long percentage = perfStatus.getGarbageCollectionDuration() * 100 / Math.max(1, status.getProcessingNanos());
                    if (percentage == 0 && perfStatus.getGarbageCollectionDuration() > 0) {
                        // If the value is non-zero but less than 1%, we want to return 1%.
                        return 1L;
                    }
                    return percentage;
                },
            processingPercentage(GARBAGE_COLLECTION_MILLIS.getDescriptor()),
            true
    );

    private static long nanosToMillis(final ProcessorStatus procStatus, final Function<ProcessingPerformanceStatus, Long> metricTransform) {
        final ProcessingPerformanceStatus perfStatus = procStatus.getProcessingPerformanceStatus();
        if (perfStatus == null) {
            return 0;
        }

        final long nanos = metricTransform.apply(perfStatus);
        final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS);
        if (millis == 0 && nanos > 0) {
            // If the value is non-zero but less than 1ms, we want to return 1ms.
            return 1;
        }

        return millis;
    }

    private static ValueReducer<StatusSnapshot, Long> processingPercentage(final MetricDescriptor<?> metricDescriptor) {
        return new ValueReducer<>() {
            @Override
            public Long reduce(final List<StatusSnapshot> values) {
                long procNanos = 0L;
                long metricMillis = 0L;

                for (final StatusSnapshot snapshot : values) {
                    final long millis = snapshot.getStatusMetric(metricDescriptor);
                    metricMillis += millis;

                    final long taskNanos = snapshot.getStatusMetric(ProcessorStatusDescriptor.TASK_NANOS.getDescriptor());
                    procNanos += taskNanos;
                }

                if (procNanos == 0) {
                    return 0L;
                }

                final long procMillis = TimeUnit.MILLISECONDS.convert(procNanos, TimeUnit.NANOSECONDS);
                // Convert processing milliseconds to 1 if nanoseconds is 0 to avoid ArithmeticException
                final long convertedProcessingMilliseconds = procMillis == 0 ? 1 : procMillis;

                return metricMillis * 100 / convertedProcessingMilliseconds;
            }
        };
    }

    private static long nanosValue(final ProcessorStatus procStatus, final Function<ProcessingPerformanceStatus, Long> metricTransform) {
        final ProcessingPerformanceStatus perfStatus = procStatus.getProcessingPerformanceStatus();
        if (perfStatus == null) {
            return 0;
        }

        return metricTransform.apply(perfStatus);
    }

    private final MetricDescriptor<ProcessorStatus> descriptor;
    private final boolean visible;

    ProcessorStatusDescriptor(final String field, final String label, final String description,
                              final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction) {

        this(field, label, description, formatter, valueFunction, true);
    }

    ProcessorStatusDescriptor(final String field, final String label, final String description,
                              final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction, final boolean visible) {

        this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction);
        this.visible = visible;
    }

    ProcessorStatusDescriptor(final String field, final String label, final String description,
                              final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction,
                              final ValueReducer<StatusSnapshot, Long> reducer, final boolean visible) {

        this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction, reducer);
        this.visible = visible;
    }

    public String getField() {
        return descriptor.getField();
    }

    public MetricDescriptor<ProcessorStatus> getDescriptor() {
        return descriptor;
    }

    public boolean isVisible() {
        return visible;
    }
}
