<commit_before>/**
 * Copyright 2017 Pivotal Software, Inc.
 * <p>
 * Licensed 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 io.micrometer.signalfx;

import com.signalfx.endpoint.SignalFxEndpoint;
import com.signalfx.endpoint.SignalFxReceiverEndpoint;
import com.signalfx.metrics.auth.StaticAuthToken;
import com.signalfx.metrics.connection.HttpDataPointProtobufReceiverFactory;
import com.signalfx.metrics.connection.HttpEventProtobufReceiverFactory;
import com.signalfx.metrics.flush.AggregateMetricSender;
import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers;
import io.micrometer.core.instrument.*;
import io.micrometer.core.instrument.config.NamingConvention;
import io.micrometer.core.instrument.step.StepMeterRegistry;
import io.micrometer.core.instrument.util.DoubleFormat;
import io.micrometer.core.instrument.util.MeterPartition;
import io.micrometer.core.lang.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.COUNTER;
import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.GAUGE;

/**
 * q
 *
 * @author <NAME>
 */
public class SignalFxMeterRegistry extends StepMeterRegistry {
    private final Logger logger = LoggerFactory.getLogger(SignalFxMeterRegistry.class);
    private final SignalFxConfig config;
    private final AggregateMetricSender metricSender;

    public SignalFxMeterRegistry(SignalFxConfig config, Clock clock) {
        this(config, clock, Executors.defaultThreadFactory());
    }

    public SignalFxMeterRegistry(SignalFxConfig config, Clock clock, ThreadFactory threadFactory) {
        super(config, clock);
        this.config = config;

        URI apiUri = URI.create(config.uri());
        int port = apiUri.getPort();
        if (port == -1) {
            if ("http".equals(apiUri.getScheme())) {
                port = 80;
            } else if ("https".equals(apiUri.getScheme())) {
                port = 443;
            }
        }

        SignalFxReceiverEndpoint signalFxEndpoint = new SignalFxEndpoint(apiUri.getScheme(), apiUri.getHost(), port);

        metricSender = new AggregateMetricSender("test.SendMetrics",
            new HttpDataPointProtobufReceiverFactory(signalFxEndpoint).setVersion(2),
            new HttpEventProtobufReceiverFactory(signalFxEndpoint),
            new StaticAuthToken(config.accessToken()),
            Collections.singleton(metricError -> logger.warn("failed to send metrics: " + metricError.getMessage())));

        config().namingConvention(new SignalFxNamingConvention());

        start(threadFactory);
    }

    @Override
    protected void publish() {
        final long timestamp = clock.wallTime();

        for (List<Meter> batch : MeterPartition.partition(this, config.batchSize())) {
            try (AggregateMetricSender.Session session = metricSender.createSession()) {
                for (Meter meter : batch) {
                    if (meter instanceof Counter) {
                        addCounter((Counter) meter, session, timestamp);
                    } else if (meter instanceof Timer) {
                        addTimer((Timer) meter, session, timestamp);
                    } else if (meter instanceof DistributionSummary) {
                        addDistributionSummary((DistributionSummary) meter, session, timestamp);
                    } else if (meter instanceof TimeGauge) {
                        addTimeGauge((TimeGauge) meter, session, timestamp);
                    } else if (meter instanceof Gauge) {
                        addGauge((Gauge) meter, session, timestamp);
                    } else if (meter instanceof FunctionTimer) {
                        addFunctionTimer((FunctionTimer) meter, session, timestamp);
                    } else if (meter instanceof FunctionCounter) {
                        addFunctionCounter((FunctionCounter) meter, session, timestamp);
                    } else if (meter instanceof LongTaskTimer) {
                        addLongTaskTimer((LongTaskTimer) meter, session, timestamp);
                    } else {
                        addMeter(meter, session, timestamp);
                    }
                }

                logger.info("successfully sent " + batch.size() + " metrics to Signalfx");
            } catch (Throwable e) {
                logger.warn("failed to send metrics", e);
            }
        }
    }

    private void addMeter(Meter meter, AggregateMetricSender.Session session, long timestamp) {
        for (Measurement measurement : meter.measure()) {
            String statSuffix = NamingConvention.camelCase.tagKey(measurement.getStatistic().toString());

            switch (measurement.getStatistic()) {
                case TOTAL:
                case TOTAL_TIME:
                case COUNT:<commit_msg>Nit: change Signalfx to SignalFx in log statement<commit_after>/**
 * Copyright 2017 Pivotal Software, Inc.
 * <p>
 * Licensed 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 io.micrometer.signalfx;

import com.signalfx.endpoint.SignalFxEndpoint;
import com.signalfx.endpoint.SignalFxReceiverEndpoint;
import com.signalfx.metrics.auth.StaticAuthToken;
import com.signalfx.metrics.connection.HttpDataPointProtobufReceiverFactory;
import com.signalfx.metrics.connection.HttpEventProtobufReceiverFactory;
import com.signalfx.metrics.flush.AggregateMetricSender;
import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers;
import io.micrometer.core.instrument.*;
import io.micrometer.core.instrument.config.NamingConvention;
import io.micrometer.core.instrument.step.StepMeterRegistry;
import io.micrometer.core.instrument.util.DoubleFormat;
import io.micrometer.core.instrument.util.MeterPartition;
import io.micrometer.core.lang.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.COUNTER;
import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.GAUGE;

/**
 * q
 *
 * @author <NAME>
 */
public class SignalFxMeterRegistry extends StepMeterRegistry {
    private final Logger logger = LoggerFactory.getLogger(SignalFxMeterRegistry.class);
    private final SignalFxConfig config;
    private final AggregateMetricSender metricSender;

    public SignalFxMeterRegistry(SignalFxConfig config, Clock clock) {
        this(config, clock, Executors.defaultThreadFactory());
    }

    public SignalFxMeterRegistry(SignalFxConfig config, Clock clock, ThreadFactory threadFactory) {
        super(config, clock);
        this.config = config;

        URI apiUri = URI.create(config.uri());
        int port = apiUri.getPort();
        if (port == -1) {
            if ("http".equals(apiUri.getScheme())) {
                port = 80;
            } else if ("https".equals(apiUri.getScheme())) {
                port = 443;
            }
        }

        SignalFxReceiverEndpoint signalFxEndpoint = new SignalFxEndpoint(apiUri.getScheme(), apiUri.getHost(), port);

        metricSender = new AggregateMetricSender("test.SendMetrics",
            new HttpDataPointProtobufReceiverFactory(signalFxEndpoint).setVersion(2),
            new HttpEventProtobufReceiverFactory(signalFxEndpoint),
            new StaticAuthToken(config.accessToken()),
            Collections.singleton(metricError -> logger.warn("failed to send metrics: " + metricError.getMessage())));

        config().namingConvention(new SignalFxNamingConvention());

        start(threadFactory);
    }

    @Override
    protected void publish() {
        final long timestamp = clock.wallTime();

        for (List<Meter> batch : MeterPartition.partition(this, config.batchSize())) {
            try (AggregateMetricSender.Session session = metricSender.createSession()) {
                for (Meter meter : batch) {
                    if (meter instanceof Counter) {
                        addCounter((Counter) meter, session, timestamp);
                    } else if (meter instanceof Timer) {
                        addTimer((Timer) meter, session, timestamp);
                    } else if (meter instanceof DistributionSummary) {
                        addDistributionSummary((DistributionSummary) meter, session, timestamp);
                    } else if (meter instanceof TimeGauge) {
                        addTimeGauge((TimeGauge) meter, session, timestamp);
                    } else if (meter instanceof Gauge) {
                        addGauge((Gauge) meter, session, timestamp);
                    } else if (meter instanceof FunctionTimer) {
                        addFunctionTimer((FunctionTimer) meter, session, timestamp);
                    } else if (meter instanceof FunctionCounter) {
                        addFunctionCounter((FunctionCounter) meter, session, timestamp);
                    } else if (meter instanceof LongTaskTimer) {
                        addLongTaskTimer((LongTaskTimer) meter, session, timestamp);
                    } else {
                        addMeter(meter, session, timestamp);
                    }
                }

                logger.info("successfully sent " + batch.size() + " metrics to SignalFx");
            } catch (Throwable e) {
                logger.warn("failed to send metrics", e);
            }
        }
    }

    private void addMeter(Meter meter, AggregateMetricSender.Session session, long timestamp) {
        for (Measurement measurement : meter.measure()) {
            String statSuffix = NamingConvention.camelCase.tagKey(measurement.getStatistic().toString());

            switch (measurement.getStatistic()) {
                case TOTAL:
                case TOTAL_TIME:
                case COUNT:
{'fraction_non_alphanumeric': 0.09184772849703716, 'fraction_numerical': 0.0030526126773208835, 'mean_word_length': 2.7606347062795407, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2376819', 'n_tokens_mistral': 2954, 'n_tokens_neox': 2793, 'n_words': 710}