<commit_before>package com.wavefront.agent;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import com.fasterxml.jackson.databind.JsonNode;
import com.squareup.tape.FileObjectQueue;
import com.squareup.tape.TaskInjector;
import com.squareup.tape.TaskQueue;
import com.wavefront.agent.api.ForceQueueEnabledAgentAPI;
import com.wavefront.api.AgentAPI;
import com.wavefront.api.agent.AgentConfiguration;
import com.wavefront.api.agent.ShellOutputDTO;
import com.wavefront.metrics.ExpectedAgentMetric;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Histogram;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.core.MetricsRegistry;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import javax.annotation.Nullable;
import javax.ws.rs.core.Response;

import static com.google.common.collect.ImmutableList.of;

/**
 * A wrapper for any AgentAPI that queues up any result posting if the backend is not available. Current data
 * will always be submitted right away (thus prioritizing live data) while background threads will submit
 * backlogged data.
 *
 * @author <NAME> (<EMAIL>)
 */
public class QueuedAgentService implements ForceQueueEnabledAgentAPI {

  private static final Logger logger = Logger.getLogger(QueuedAgentService.class.getCanonicalName());

  private final Gson resubmissionTaskMarshaller;
  private final AgentAPI wrapped;
  private final UUID agentId;
  private final List<TaskQueue<ResubmissionTask>> taskQueues;
  private boolean lastKnownQueueSizeIsPositive = true;
  private MetricsRegistry metricsRegistry = new MetricsRegistry();
  private Meter resultPostingMeter = metricsRegistry.newMeter(QueuedAgentService.class, "post-result", "results",
      TimeUnit.MINUTES);
  /**
   * Biases result sizes to the last 5 minutes heavily. This histogram does not see all result sizes. The executor
   * only ever processes one posting at any given time and drops the rest. {@link #resultPostingMeter} records the
   * actual rate (i.e. sees all posting calls).
   */
  private Histogram resultPostingSizes = metricsRegistry.newHistogram(QueuedAgentService.class, "result-size", true);
  /**
   * A single threaded bounded work queue to update result posting sizes.
   */
  private ExecutorService resultPostingSizerExecutorService = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS,
      new ArrayBlockingQueue<Runnable>(1));

  /**
   * @return bytes per minute for requests submissions. Null if no data is available yet.
   */
  @Nullable
  public Long getBytesPerMinute() {
    if (resultPostingMeter.fifteenMinuteRate() == 0 || resultPostingSizes.mean() == 0 || resultPostingSizes.count() <
        50) {
      return null;
    }
    return (long) (resultPostingSizes.mean() * resultPostingMeter.fifteenMinuteRate());
  }

  public QueuedAgentService(AgentAPI service, String bufferFile, int retryThreads,
                            ScheduledExecutorService executorService, boolean purge, final UUID agentId)
      throws IOException {
    this.agentId = agentId;
    if (retryThreads <= 0) {
      logger.warning("You have no retry threads set up. Any points that get rejected will be lost.\n Change this by " +
          "setting retryThreads to a value > 0");
    }
    resubmissionTaskMarshaller = new GsonBuilder().
        registerTypeHierarchyAdapter(ResubmissionTask.class, new ResubmissionTaskDeserializer()).create();
    this.wrapped = service;
    this.taskQueues = Lists.newArrayListWithExpectedSize(retryThreads);
    for (int i = 0; i < retryThreads; i++) {
      final int threadId = i;
      File buffer = new File(bufferFile + "." + i);
      if (purge) {
        if (buffer.delete()) {
          logger.warning("Retry buffer has been purged: " + buffer.getAbsolutePath());
        }
      }
      final TaskQueue<ResubmissionTask> taskQueue = new TaskQueue<>(
          new FileObjectQueue<>(buffer, new FileObjectQueue.Converter<ResubmissionTask>() {
            @Override
            public ResubmissionTask from(byte[] bytes) throws IOException {
              try {
                Reader reader = new InputStreamReader(new GZIPInputStream(new ByteArrayInputStream(bytes)));<commit_msg>clean up<commit_after>package com.wavefront.agent;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import com.fasterxml.jackson.databind.JsonNode;
import com.squareup.tape.FileObjectQueue;
import com.squareup.tape.TaskInjector;
import com.squareup.tape.TaskQueue;
import com.wavefront.agent.api.ForceQueueEnabledAgentAPI;
import com.wavefront.api.AgentAPI;
import com.wavefront.api.agent.AgentConfiguration;
import com.wavefront.api.agent.ShellOutputDTO;
import com.wavefront.metrics.ExpectedAgentMetric;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Histogram;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.core.MetricsRegistry;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import javax.annotation.Nullable;
import javax.ws.rs.core.Response;

import static com.google.common.collect.ImmutableList.of;

/**
 * A wrapper for any AgentAPI that queues up any result posting if the backend is not available. Current data
 * will always be submitted right away (thus prioritizing live data) while background threads will submit
 * backlogged data.
 *
 * @author <NAME> (<EMAIL>)
 */
public class QueuedAgentService implements ForceQueueEnabledAgentAPI {

  private static final Logger logger = Logger.getLogger(QueuedAgentService.class.getCanonicalName());

  private final Gson resubmissionTaskMarshaller;
  private final AgentAPI wrapped;
  private final List<TaskQueue<ResubmissionTask>> taskQueues;
  private boolean lastKnownQueueSizeIsPositive = true;
  private MetricsRegistry metricsRegistry = new MetricsRegistry();
  private Meter resultPostingMeter = metricsRegistry.newMeter(QueuedAgentService.class, "post-result", "results",
      TimeUnit.MINUTES);
  /**
   * Biases result sizes to the last 5 minutes heavily. This histogram does not see all result sizes. The executor
   * only ever processes one posting at any given time and drops the rest. {@link #resultPostingMeter} records the
   * actual rate (i.e. sees all posting calls).
   */
  private Histogram resultPostingSizes = metricsRegistry.newHistogram(QueuedAgentService.class, "result-size", true);
  /**
   * A single threaded bounded work queue to update result posting sizes.
   */
  private ExecutorService resultPostingSizerExecutorService = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS,
      new ArrayBlockingQueue<Runnable>(1));

  /**
   * @return bytes per minute for requests submissions. Null if no data is available yet.
   */
  @Nullable
  public Long getBytesPerMinute() {
    if (resultPostingMeter.fifteenMinuteRate() == 0 || resultPostingSizes.mean() == 0 || resultPostingSizes.count() <
        50) {
      return null;
    }
    return (long) (resultPostingSizes.mean() * resultPostingMeter.fifteenMinuteRate());
  }

  public QueuedAgentService(AgentAPI service, String bufferFile, int retryThreads,
                            ScheduledExecutorService executorService, boolean purge, final UUID agentId)
      throws IOException {
    if (retryThreads <= 0) {
      logger.warning("You have no retry threads set up. Any points that get rejected will be lost.\n Change this by " +
          "setting retryThreads to a value > 0");
    }
    resubmissionTaskMarshaller = new GsonBuilder().
        registerTypeHierarchyAdapter(ResubmissionTask.class, new ResubmissionTaskDeserializer()).create();
    this.wrapped = service;
    this.taskQueues = Lists.newArrayListWithExpectedSize(retryThreads);
    for (int i = 0; i < retryThreads; i++) {
      final int threadId = i;
      File buffer = new File(bufferFile + "." + i);
      if (purge) {
        if (buffer.delete()) {
          logger.warning("Retry buffer has been purged: " + buffer.getAbsolutePath());
        }
      }
      final TaskQueue<ResubmissionTask> taskQueue = new TaskQueue<>(
          new FileObjectQueue<>(buffer, new FileObjectQueue.Converter<ResubmissionTask>() {
            @Override
            public ResubmissionTask from(byte[] bytes) throws IOException {
              try {
                Reader reader = new InputStreamReader(new GZIPInputStream(new ByteArrayInputStream(bytes)));
{'fraction_non_alphanumeric': 0.09337779895188185, 'fraction_numerical': 0.0024773701762744165, 'mean_word_length': 4.421487603305785, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '2495874', 'n_tokens_mistral': 2997, 'n_tokens_neox': 2958, 'n_words': 770}