/*
 * 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.reporting.prometheus.api;

import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import io.prometheus.client.SimpleCollector;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.status.ConnectionStatus;
import org.apache.nifi.controller.status.PortStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;

import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Gauge;
import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
import org.apache.nifi.controller.status.TransmissionStatus;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.metrics.jvm.JvmMetrics;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.ssl.RestrictedSSLContextService;

public class PrometheusMetricsUtil {

    public static final AllowableValue METRICS_STRATEGY_ROOT = new AllowableValue("Root Process Group", "Root Process Group",
            "Send rollup metrics for the entire root process group");
    public static final AllowableValue METRICS_STRATEGY_PG = new AllowableValue("All Process Groups", "All Process Groups",
            "Send metrics for each process group");
    public static final AllowableValue METRICS_STRATEGY_COMPONENTS = new AllowableValue("All Components", "All Components",
            "Send metrics for each component in the system, to include processors, connections, controller services, etc.");

    private static final CollectorRegistry NIFI_REGISTRY = new CollectorRegistry();
    private static final CollectorRegistry JVM_REGISTRY = new CollectorRegistry();

    // Common properties/values
    public static final AllowableValue CLIENT_NONE = new AllowableValue("No Authentication", "No Authentication",
            "ReportingTask will not authenticate clients. Anyone can communicate with this ReportingTask anonymously");
    public static final AllowableValue CLIENT_WANT = new AllowableValue("Want Authentication", "Want Authentication",
            "ReportingTask will try to verify the client but if unable to verify will allow the client to communicate anonymously");
    public static final AllowableValue CLIENT_NEED = new AllowableValue("Need Authentication", "Need Authentication",
            "ReportingTask will reject communications from any client unless the client provides a certificate that is trusted by the TrustStore"
                    + "specified in the SSL Context Service");

    public static final PropertyDescriptor METRICS_ENDPOINT_PORT = new PropertyDescriptor.Builder()
            .name("prometheus-reporting-task-metrics-endpoint-port")
            .displayName("Prometheus Metrics Endpoint Port")
            .description("The Port where prometheus metrics can be accessed")
            .required(true)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .defaultValue("9092")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .build();

    public static final PropertyDescriptor INSTANCE_ID = new PropertyDescriptor.Builder()
            .name("prometheus-reporting-task-instance-id")
            .displayName("Instance ID")
            .description("Id of this NiFi instance to be included in the metrics sent to Prometheus")
            .required(true)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .defaultValue("${hostname(true)}")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor SSL_CONTEXT = new PropertyDescriptor.Builder()
            .name("prometheus-reporting-task-ssl-context")
            .displayName("SSL Context Service")
            .description("The SSL Context Service to use in order to secure the server. If specified, the server will"
                    + "accept only HTTPS requests; otherwise, the server will accept only HTTP requests")
            .required(false)
            .identifiesControllerService(RestrictedSSLContextService.class)
            .build();

    public static final PropertyDescriptor CLIENT_AUTH = new PropertyDescriptor.Builder()
            .name("prometheus-reporting-task-client-auth")
            .displayName("Client Authentication")
            .description("Specifies whether or not the Reporting Task should authenticate clients. This value is ignored if the <SSL Context Service> "
                    + "Property is not specified or the SSL Context provided uses only a KeyStore and not a TrustStore.")
            .required(true)
            .allowableValues(CLIENT_NONE, CLIENT_WANT, CLIENT_NEED)
            .defaultValue(CLIENT_NONE.getValue())
            .build();

    // Processor / Process Group metrics
    private static final Gauge AMOUNT_FLOWFILES_SENT = Gauge.build()
            .name("nifi_amount_flowfiles_sent")
            .help("Total number of FlowFiles sent by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_FLOWFILES_TRANSFERRED = Gauge.build()
            .name("nifi_amount_flowfiles_transferred")
            .help("Total number of FlowFiles transferred by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_FLOWFILES_RECEIVED = Gauge.build()
            .name("nifi_amount_flowfiles_received")
            .help("Total number of FlowFiles received by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_FLOWFILES_REMOVED = Gauge.build()
            .name("nifi_amount_flowfiles_removed")
            .help("Total number of FlowFiles removed by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_BYTES_SENT = Gauge.build()
            .name("nifi_amount_bytes_sent")
            .help("Total number of bytes sent by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_BYTES_READ = Gauge.build()
            .name("nifi_amount_bytes_read")
            .help("Total number of bytes read by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_BYTES_WRITTEN = Gauge.build()
            .name("nifi_amount_bytes_written")
            .help("Total number of bytes written by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_BYTES_RECEIVED = Gauge.build()
            .name("nifi_amount_bytes_received")
            .help("Total number of bytes received by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_BYTES_TRANSFERRED = Gauge.build()
            .name("nifi_amount_bytes_transferred")
            .help("Total number of Bytes transferred by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_THREADS_TOTAL_ACTIVE = Gauge.build()
            .name("nifi_amount_threads_active")
            .help("Total number of threads active for the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_THREADS_TOTAL_TERMINATED = Gauge.build()
            .name("nifi_amount_threads_terminated")
            .help("Total number of threads terminated for the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id")
            .register(NIFI_REGISTRY);

    private static final Gauge SIZE_CONTENT_OUTPUT_TOTAL = Gauge.build()
            .name("nifi_size_content_output_total")
            .help("Total size of content output by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge SIZE_CONTENT_INPUT_TOTAL = Gauge.build()
            .name("nifi_size_content_input_total")
            .help("Total size of content input by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge SIZE_CONTENT_QUEUED_TOTAL = Gauge.build()
            .name("nifi_size_content_queued_total")
            .help("Total size of content queued in the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_ITEMS_OUTPUT = Gauge.build()
            .name("nifi_amount_items_output")
            .help("Total number of items output by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_ITEMS_INPUT = Gauge.build()
            .name("nifi_amount_items_input")
            .help("Total number of items input by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge AMOUNT_ITEMS_QUEUED = Gauge.build()
            .name("nifi_amount_items_queued")
            .help("Total number of items queued by the component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    // Processor metrics
    private static final Gauge PROCESSOR_COUNTERS = Gauge.build()
            .name("nifi_processor_counters")
            .help("Counters exposed by NiFi Processors")
            .labelNames("processor_name", "counter_name", "processor_id", "instance")
            .register(NIFI_REGISTRY);

    // Connection metrics
    private static final Gauge BACKPRESSURE_BYTES_THRESHOLD = Gauge.build()
            .name("nifi_backpressure_bytes_threshold")
            .help("The number of bytes that can be queued before backpressure is applied")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge BACKPRESSURE_OBJECT_THRESHOLD = Gauge.build()
            .name("nifi_backpressure_object_threshold")
            .help("The number of flow files that can be queued before backpressure is applied")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge IS_BACKPRESSURE_ENABLED = Gauge.build()
            .name("nifi_backpressure_enabled")
            .help("Whether backpressure has been applied for this component. Values are 0 or 1")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    // Port metrics
    private static final Gauge IS_TRANSMITTING = Gauge.build()
            .name("nifi_transmitting")
            .help("Whether this component is transmitting data. Values are 0 or 1")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id", "run_status")
            .register(NIFI_REGISTRY);

    // Remote Process Group (RPG) metrics
    private static final Gauge ACTIVE_REMOTE_PORT_COUNT = Gauge.build()
            .name("nifi_active_remote_port_count")
            .help("The number of active remote ports associated with this component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge INACTIVE_REMOTE_PORT_COUNT = Gauge.build()
            .name("nifi_inactive_remote_port_count")
            .help("The number of inactive remote ports associated with this component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    private static final Gauge AVERAGE_LINEAGE_DURATION = Gauge.build()
            .name("nifi_average_lineage_duration")
            .help("The average lineage duration (in milliseconds) for all flow file processed by this component")
            .labelNames("instance", "component_type", "component_name", "component_id", "parent_id",
                    "source_id", "source_name", "destination_id", "destination_name")
            .register(NIFI_REGISTRY);

    ///////////////////////////////////////////////////////////////
    // JVM Metrics
    ///////////////////////////////////////////////////////////////
    private static final Gauge JVM_HEAP_USED = Gauge.build()
            .name("nifi_jvm_heap_used")
            .help("NiFi JVM heap used")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_HEAP_USAGE = Gauge.build()
            .name("nifi_jvm_heap_usage")
            .help("NiFi JVM heap usage")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_HEAP_NON_USAGE = Gauge.build()
            .name("nifi_jvm_heap_non_usage")
            .help("NiFi JVM heap non usage")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_THREAD_COUNT = Gauge.build()
            .name("nifi_jvm_thread_count")
            .help("NiFi JVM thread count")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_DAEMON_THREAD_COUNT = Gauge.build()
            .name("nifi_jvm_daemon_thread_count")
            .help("NiFi JVM daemon thread count")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_UPTIME = Gauge.build()
            .name("nifi_jvm_uptime")
            .help("NiFi JVM uptime")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_FILE_DESCRIPTOR_USAGE = Gauge.build()
            .name("nifi_jvm_file_descriptor_usage")
            .help("NiFi JVM file descriptor usage")
            .labelNames("instance")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_GC_RUNS = Gauge.build()
            .name("nifi_jvm_gc_runs")
            .help("NiFi JVM GC number of runs")
            .labelNames("instance", "gc_name")
            .register(JVM_REGISTRY);

    private static final Gauge JVM_GC_TIME = Gauge.build()
            .name("nifi_jvm_gc_time")
            .help("NiFi JVM GC time in milliseconds")
            .labelNames("instance", "gc_name")
            .register(JVM_REGISTRY);

    public static CollectorRegistry createNifiMetrics(ProcessGroupStatus status, String instanceId, String parentPGId, String componentType, String metricsStrategy) {

        final String componentId = status.getId();
        final String componentName = status.getName();

        // Clear all collectors to deal with removed/renamed components -- for root PG only
        if("RootProcessGroup".equals(componentType)) {
            try {
                for (final Field field : PrometheusMetricsUtil.class.getDeclaredFields()) {
                    if (Modifier.isStatic(field.getModifiers()) && (field.get(null) instanceof SimpleCollector)) {
                        SimpleCollector<?> sc = (SimpleCollector<?>) (field.get(null));
                        sc.clear();
                    }
                }
            } catch (IllegalAccessException e) {
                // ignore
            }
        }

        AMOUNT_FLOWFILES_SENT.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getFlowFilesSent());
        AMOUNT_FLOWFILES_TRANSFERRED.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getFlowFilesTransferred());
        AMOUNT_FLOWFILES_RECEIVED.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getFlowFilesReceived());

        AMOUNT_BYTES_SENT.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getBytesSent());
        AMOUNT_BYTES_READ.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getBytesRead());
        AMOUNT_BYTES_WRITTEN.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getBytesWritten());
        AMOUNT_BYTES_RECEIVED.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getBytesReceived());
        AMOUNT_BYTES_TRANSFERRED.labels(instanceId, componentType, componentName, componentId, parentPGId).set(status.getBytesTransferred());

        SIZE_CONTENT_OUTPUT_TOTAL.labels(instanceId, componentType, componentName, componentId, parentPGId, "", "", "", "")
                .set(status.getOutputContentSize());
        SIZE_CONTENT_INPUT_TOTAL.labels(instanceId, componentType, componentName, componentId, parentPGId, "", "", "", "")
                .set(status.getInputContentSize());
        SIZE_CONTENT_QUEUED_TOTAL.labels(instanceId, componentType, componentName, componentId, parentPGId, "", "", "", "")
                .set(status.getQueuedContentSize());

        AMOUNT_ITEMS_OUTPUT.labels(instanceId, componentType, componentName, componentId, parentPGId, "", "", "", "")
                .set(status.getOutputCount());
        AMOUNT_ITEMS_INPUT.labels(instanceId, componentType, componentName, componentId, parentPGId, "", "", "", "")
                .set(status.getInputCount());
        AMOUNT_ITEMS_QUEUED.labels(instanceId, componentType, componentName, componentId, parentPGId,"", "", "", "")
                .set(status.getQueuedCount());

        AMOUNT_THREADS_TOTAL_ACTIVE.labels(instanceId, componentType, componentName, componentId, parentPGId)
                .set(status.getActiveThreadCount() == null ? 0 : status.getActiveThreadCount());
        AMOUNT_THREADS_TOTAL_TERMINATED.labels(instanceId, componentType, componentName, componentId, parentPGId)
                .set(status.getTerminatedThreadCount() == null ? 0 : status.getTerminatedThreadCount());

        // Report metrics for child process groups if specified
        if (METRICS_STRATEGY_PG.getValue().equals(metricsStrategy) || METRICS_STRATEGY_COMPONENTS.getValue().equals(metricsStrategy)) {
            status.getProcessGroupStatus().forEach((childGroupStatus) -> createNifiMetrics(childGroupStatus, instanceId, componentId, "ProcessGroup", metricsStrategy));
        }

        if (METRICS_STRATEGY_COMPONENTS.getValue().equals(metricsStrategy)) {
            // Report metrics for all components
            for (ProcessorStatus processorStatus : status.getProcessorStatus()) {
                Map<String, Long> counters = processorStatus.getCounters();

                if (counters != null) {
                    counters.entrySet().stream().forEach(entry -> PROCESSOR_COUNTERS
                            .labels(processorStatus.getName(), entry.getKey(), processorStatus.getId(), instanceId).set(entry.getValue()));
                }

                final String procComponentType = "Processor";
                final String procComponentId = processorStatus.getId();
                final String procComponentName = processorStatus.getName();
                final String parentId = processorStatus.getGroupId();

                AMOUNT_FLOWFILES_SENT.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getFlowFilesSent());
                AMOUNT_FLOWFILES_RECEIVED.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getFlowFilesReceived());
                AMOUNT_FLOWFILES_REMOVED.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getFlowFilesRemoved());

                AMOUNT_BYTES_SENT.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getBytesSent());
                AMOUNT_BYTES_READ.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getBytesRead());
                AMOUNT_BYTES_WRITTEN.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getBytesWritten());
                AMOUNT_BYTES_RECEIVED.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId).set(processorStatus.getBytesReceived());

                SIZE_CONTENT_OUTPUT_TOTAL.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId, "", "", "", "")
                        .set(processorStatus.getOutputBytes());
                SIZE_CONTENT_INPUT_TOTAL.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId, "", "", "", "")
                        .set(processorStatus.getInputBytes());

                AMOUNT_ITEMS_OUTPUT.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId, "", "", "", "")
                        .set(processorStatus.getOutputCount());
                AMOUNT_ITEMS_INPUT.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId, "", "", "", "")
                        .set(processorStatus.getInputCount());

                AVERAGE_LINEAGE_DURATION.labels(instanceId, procComponentType, procComponentName, procComponentId, parentPGId, "", "", "", "")
                        .set(processorStatus.getAverageLineageDuration());

                AMOUNT_THREADS_TOTAL_ACTIVE.labels(instanceId, procComponentType, procComponentName, procComponentId, parentId)
                        .set(status.getActiveThreadCount() == null ? 0 : status.getActiveThreadCount());
                AMOUNT_THREADS_TOTAL_TERMINATED.labels(instanceId, procComponentType, procComponentName, procComponentId, parentId)
                        .set(status.getTerminatedThreadCount() == null ? 0 : status.getTerminatedThreadCount());

            }
            for (ConnectionStatus connectionStatus : status.getConnectionStatus()) {
                final String connComponentId = connectionStatus.getId();
                final String connComponentName = connectionStatus.getName();
                final String sourceId = connectionStatus.getSourceId();
                final String sourceName = connectionStatus.getSourceName();
                final String destinationId = connectionStatus.getDestinationId();
                final String destinationName = connectionStatus.getDestinationName();
                final String parentId = connectionStatus.getGroupId();
                final String connComponentType = "Connection";
                SIZE_CONTENT_OUTPUT_TOTAL.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getOutputBytes());
                SIZE_CONTENT_INPUT_TOTAL.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getInputBytes());
                SIZE_CONTENT_QUEUED_TOTAL.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getQueuedBytes());

                AMOUNT_ITEMS_OUTPUT.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getOutputCount());
                AMOUNT_ITEMS_INPUT.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getInputCount());
                AMOUNT_ITEMS_QUEUED.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getQueuedCount());

                BACKPRESSURE_BYTES_THRESHOLD.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getBackPressureBytesThreshold());
                BACKPRESSURE_OBJECT_THRESHOLD.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(connectionStatus.getBackPressureObjectThreshold());
                boolean isBackpressureEnabled = (connectionStatus.getBackPressureObjectThreshold() > 0 && connectionStatus.getBackPressureObjectThreshold() <= connectionStatus.getQueuedCount())
                        || (connectionStatus.getBackPressureBytesThreshold() > 0 && connectionStatus.getBackPressureBytesThreshold() <= connectionStatus.getMaxQueuedBytes());
                IS_BACKPRESSURE_ENABLED.labels(instanceId, connComponentType, connComponentName, connComponentId, parentId, sourceId, sourceName, destinationId, destinationName)
                        .set(isBackpressureEnabled ? 1 : 0);
            }
            for (PortStatus portStatus : status.getInputPortStatus()) {
                final String portComponentId = portStatus.getId();
                final String portComponentName = portStatus.getName();
                final String parentId = portStatus.getGroupId();
                final String portComponentType = "InputPort";
                AMOUNT_FLOWFILES_SENT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getFlowFilesSent());
                AMOUNT_FLOWFILES_RECEIVED.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getFlowFilesReceived());

                AMOUNT_BYTES_SENT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getBytesSent());
                AMOUNT_BYTES_READ.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getInputBytes());
                AMOUNT_BYTES_WRITTEN.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getOutputBytes());
                AMOUNT_BYTES_RECEIVED.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getBytesReceived());

                AMOUNT_ITEMS_OUTPUT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, "", "", "", "")
                        .set(portStatus.getOutputCount());
                AMOUNT_ITEMS_INPUT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, "", "", "", "")
                        .set(portStatus.getInputCount());

                final Boolean isTransmitting = portStatus.isTransmitting();
                IS_TRANSMITTING.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, portStatus.getRunStatus().name())
                        .set(isTransmitting == null ? 0 : (isTransmitting ? 1 : 0));

                AMOUNT_THREADS_TOTAL_ACTIVE.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getActiveThreadCount());
            }
            for (PortStatus portStatus : status.getOutputPortStatus()) {
                final String portComponentId = portStatus.getId();
                final String portComponentName = portStatus.getName();
                final String parentId = portStatus.getGroupId();
                final String portComponentType = "OutputPort";
                AMOUNT_FLOWFILES_SENT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getFlowFilesSent());
                AMOUNT_FLOWFILES_RECEIVED.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getFlowFilesReceived());

                AMOUNT_BYTES_SENT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getBytesSent());
                AMOUNT_BYTES_READ.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getInputBytes());
                AMOUNT_BYTES_WRITTEN.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getOutputBytes());
                AMOUNT_BYTES_RECEIVED.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getBytesReceived());

                AMOUNT_ITEMS_OUTPUT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, "", "", "", "")
                        .set(portStatus.getOutputCount());
                AMOUNT_ITEMS_INPUT.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, "", "", "", "")
                        .set(portStatus.getInputCount());

                final Boolean isTransmitting = portStatus.isTransmitting();
                IS_TRANSMITTING.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId, portStatus.getRunStatus().name())
                        .set(isTransmitting == null ? 0 : (isTransmitting ? 1 : 0));

                AMOUNT_THREADS_TOTAL_ACTIVE.labels(instanceId, portComponentType, portComponentName, portComponentId, parentId).set(portStatus.getActiveThreadCount());
            }
            for (RemoteProcessGroupStatus remoteProcessGroupStatus : status.getRemoteProcessGroupStatus()) {
                final String rpgComponentId = remoteProcessGroupStatus.getId();
                final String rpgComponentName = remoteProcessGroupStatus.getName();
                final String parentId = remoteProcessGroupStatus.getGroupId();
                final String rpgComponentType = "RemoteProcessGroup";

                AMOUNT_BYTES_WRITTEN.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId).set(remoteProcessGroupStatus.getSentContentSize());
                AMOUNT_BYTES_RECEIVED.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId).set(remoteProcessGroupStatus.getReceivedContentSize());

                AMOUNT_ITEMS_OUTPUT.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, "", "", "", "")
                        .set(remoteProcessGroupStatus.getSentCount());
                AMOUNT_ITEMS_INPUT.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, "", "", "", "")
                        .set(remoteProcessGroupStatus.getReceivedCount());

                ACTIVE_REMOTE_PORT_COUNT.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, "", "", "", "")
                        .set(remoteProcessGroupStatus.getActiveRemotePortCount());
                INACTIVE_REMOTE_PORT_COUNT.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, "", "", "", "")
                        .set(remoteProcessGroupStatus.getInactiveRemotePortCount());

                AVERAGE_LINEAGE_DURATION.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, "", "", "", "")
                        .set(remoteProcessGroupStatus.getAverageLineageDuration());

                IS_TRANSMITTING.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId, remoteProcessGroupStatus.getTransmissionStatus().name())
                        .set(TransmissionStatus.Transmitting.equals(remoteProcessGroupStatus.getTransmissionStatus()) ? 1 : 0);

                AMOUNT_THREADS_TOTAL_ACTIVE.labels(instanceId, rpgComponentType, rpgComponentName, rpgComponentId, parentId).set(remoteProcessGroupStatus.getActiveThreadCount());
            }
        }

        return NIFI_REGISTRY;
    }

    public static CollectorRegistry createJvmMetrics(JvmMetrics jvmMetrics, String instanceId) {
        JVM_HEAP_USED.labels(instanceId).set(jvmMetrics.heapUsed(DataUnit.B));
        JVM_HEAP_USAGE.labels(instanceId).set(jvmMetrics.heapUsage());
        JVM_HEAP_NON_USAGE.labels(instanceId).set(jvmMetrics.nonHeapUsage());

        JVM_THREAD_COUNT.labels(instanceId).set(jvmMetrics.threadCount());
        JVM_DAEMON_THREAD_COUNT.labels(instanceId).set(jvmMetrics.daemonThreadCount());

        JVM_UPTIME.labels(instanceId).set(jvmMetrics.uptime());
        JVM_FILE_DESCRIPTOR_USAGE.labels(instanceId).set(jvmMetrics.fileDescriptorUsage());

        jvmMetrics.garbageCollectors()
                .forEach((name, stat) -> {
                    JVM_GC_RUNS.labels(instanceId, name).set(stat.getRuns());
                    JVM_GC_TIME.labels(instanceId, name).set(stat.getTime(TimeUnit.MILLISECONDS));
                });

        return JVM_REGISTRY;
    }

}