<commit_before>package org.apache.s4.core.util;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.s4.base.Emitter;
import org.apache.s4.comm.topology.Assignment;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.ReceiverImpl;
import org.apache.s4.core.RemoteSender;
import org.apache.s4.core.SenderImpl;
import org.apache.s4.core.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beust.jcommander.internal.Lists;
import com.beust.jcommander.internal.Maps;
import com.google.common.cache.LoadingCache;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Meter;

@Singleton
public class S4Metrics {

    private static Logger logger = LoggerFactory.getLogger(S4Metrics.class);

    @Inject
    Emitter emitter;

    @Inject
    Assignment assignment;

    static List<Meter> partitionSenderMeters = Lists.newArrayList();

    private final Meter eventMeter = Metrics.newMeter(ReceiverImpl.class, "received-events", "event-count",
            TimeUnit.SECONDS);
    private final Meter bytesMeter = Metrics.newMeter(ReceiverImpl.class, "received-bytes", "bytes-count",
            TimeUnit.SECONDS);

    private final Meter localEventsMeter = Metrics.newMeter(Stream.class, "sent-local", "sent-local", TimeUnit.SECONDS);
    private final Meter remoteEventsMeter = Metrics.newMeter(Stream.class, "sent-remote", "sent-remote",
            TimeUnit.SECONDS);

    private Meter[] senderMeters;

    private final Map<String, Meter> dequeuingStreamMeters = Maps.newHashMap();
    private final Map<String, Meter> droppedStreamMeters = Maps.newHashMap();
    private final Map<String, Meter> streamQueueFullMeters = Maps.newHashMap();
    private final Meter droppedInSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@sender", "dropped@sender",
            TimeUnit.SECONDS);
    private final Meter droppedInRemoteSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@remote-sender",
            "dropped@remote-sender", TimeUnit.SECONDS);

    private final Map<String, Meter[]> remoteSenderMeters = Maps.newHashMap();

    @Inject
    private void init() {
        senderMeters = new Meter[emitter.getPartitionCount()];
        // int localPartitionId = assignment.assignClusterNode().getPartition();
        for (int i = 0; i < senderMeters.length; i++) {
            senderMeters[i] = Metrics.newMeter(SenderImpl.class, "sender", "sent-to-" + (i), TimeUnit.SECONDS);
        }
        Metrics.newGauge(Stream.class, "local-vs-remote", new Gauge<Double>() {
            @Override
            public Double value() {
                // this will return NaN if divider is zero
                return localEventsMeter.oneMinuteRate() / remoteEventsMeter.oneMinuteRate();
            }
        });

    }

    public void createCacheGauges(ProcessingElement prototype, final LoadingCache<String, ProcessingElement> cache) {

        Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-entries", new Gauge<Long>() {

            @Override
            public Long value() {
                return cache.size();
            }
        });<commit_msg>Add metrics reporting initialization<commit_after>package org.apache.s4.core.util;

import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.s4.base.Emitter;
import org.apache.s4.comm.topology.Assignment;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.ReceiverImpl;
import org.apache.s4.core.RemoteSender;
import org.apache.s4.core.SenderImpl;
import org.apache.s4.core.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beust.jcommander.internal.Lists;
import com.beust.jcommander.internal.Maps;
import com.google.common.base.Strings;
import com.google.common.cache.LoadingCache;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.reporting.ConsoleReporter;
import com.yammer.metrics.reporting.CsvReporter;

@Singleton
public class S4Metrics {

    private static Logger logger = LoggerFactory.getLogger(S4Metrics.class);

    static final Pattern METRICS_CONFIG_PATTERN = Pattern
            .compile("(csv:.+|console):(\\d+):(DAYS|HOURS|MICROSECONDS|MILLISECONDS|MINUTES|NANOSECONDS|SECONDS)");

    @Inject
    Emitter emitter;

    @Inject
    Assignment assignment;

    @Inject(optional = true)
    @Named("s4.metrics.config")
    String metricsConfig;

    static List<Meter> partitionSenderMeters = Lists.newArrayList();

    private final Meter eventMeter = Metrics.newMeter(ReceiverImpl.class, "received-events", "event-count",
            TimeUnit.SECONDS);
    private final Meter bytesMeter = Metrics.newMeter(ReceiverImpl.class, "received-bytes", "bytes-count",
            TimeUnit.SECONDS);

    private final Meter localEventsMeter = Metrics.newMeter(Stream.class, "sent-local", "sent-local", TimeUnit.SECONDS);
    private final Meter remoteEventsMeter = Metrics.newMeter(Stream.class, "sent-remote", "sent-remote",
            TimeUnit.SECONDS);

    private Meter[] senderMeters;

    private final Map<String, Meter> dequeuingStreamMeters = Maps.newHashMap();
    private final Map<String, Meter> droppedStreamMeters = Maps.newHashMap();
    private final Map<String, Meter> streamQueueFullMeters = Maps.newHashMap();
    private final Meter droppedInSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@sender", "dropped@sender",
            TimeUnit.SECONDS);
    private final Meter droppedInRemoteSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@remote-sender",
            "dropped@remote-sender", TimeUnit.SECONDS);

    private final Map<String, Meter[]> remoteSenderMeters = Maps.newHashMap();

    @Inject
    private void init() {

        if (Strings.isNullOrEmpty(metricsConfig)) {
            logger.info("Metrics reporting not configured");
        } else {
            Matcher matcher = METRICS_CONFIG_PATTERN.matcher(metricsConfig);
            if (!matcher.matches()) {
                logger.error(
                        "Invalid metrics configuration [{}]. Metrics configuration must match the pattern [{}]. Metrics reporting disabled.",
                        metricsConfig, METRICS_CONFIG_PATTERN);
            } else {
                matcher.find();
                String group1 = matcher.group(1);

                if ("csv".equals(group1)) {
                    String outputDir = matcher.group(2);
                    long period = Long.valueOf(matcher.group(3));
                    TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(4));
                    logger.info("Reporting metrics through csv files in directory [{}] with frequency of [{}] [{}]",
                            new String[] { outputDir, String.valueOf(period), timeUnit.name() });
                    CsvReporter.enable(new File(outputDir), period, timeUnit);
                } else {
                    long period = Long.valueOf(matcher.group(2));
                    TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(3));
                    logger.info("Reporting metrics on the console with frequency of [{}] [{}]",
                            new String[] { String.valueOf(period), timeUnit.name() });
                    ConsoleReporter.enable(period, timeUnit);
                }
            }
        }

        senderMeters = new Meter[emitter.getPartitionCount()];
        // int localPartitionId = assignment.assignClusterNode().getPartition();
        for (int i = 0; i < senderMeters.length; i++) {
            senderMeters[i] = Metrics.newMeter(SenderImpl.class, "sender", "sent-to-" + (i), TimeUnit.SECONDS);
        }
        Metrics.newGauge(Stream.class, "local-vs-remote", new Gauge<Double>() {
            @Override
            public Double value() {
                // this will return NaN if divider is zero
                return localEventsMeter.oneMinuteRate() / remoteEventsMeter.oneMinuteRate();
            }
        });

    }

    public void createCacheGauges(ProcessingElement prototype, final LoadingCache<String, ProcessingElement> cache) {

        Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-entries", new Gauge<Long>() {

            @Override
            public Long value() {
                return cache.size();
            }
        });
{'fraction_non_alphanumeric': 0.11510708968044897, 'fraction_numerical': 0.004008704615737029, 'mean_word_length': 3.0632852489530014, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2406437', 'n_tokens_mistral': 2568, 'n_tokens_neox': 2512, 'n_words': 505}