<commit_before>/*
 * Copyright 2016-present Facebook, Inc.
 *
 * 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
 *
 *     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 com.facebook.buck.util.network.offline;

import com.facebook.buck.counters.CounterRegistry;
import com.facebook.buck.counters.IntegerCounter;
import com.facebook.buck.event.BuckEventBus;
import com.facebook.buck.io.ProjectFilesystem;
import com.facebook.buck.log.Logger;
import com.facebook.buck.model.BuildId;
import com.facebook.buck.util.network.ScribeLogger;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * This logger uses files-related operations (for offline logging i.e. storing and delayed logging
 * once network connection is better). Some of them may not be supported by all filesystems. E.g.
 * Jimfs uses JimfsPath which does not support toFile() operation. For such filesystems, the offline
 * logging will not be performed.
 */
public class OfflineScribeLogger extends ScribeLogger {

  protected static final String LOGFILE_PATTERN = "scribe-stored-*.log";
  protected static final String LOGFILE_PREFIX = "scribe-stored-";
  protected static final String LOGFILE_SUFFIX = ".log";

  private static final int BUFFER_SIZE = 1024 * 1024;
  private static final long CLUSTER_DISPATCH_SIZE = 1024 * 1024;
  private static final int KILO = 1024;
  private static final Logger LOG = Logger.get(OfflineScribeLogger.class);
  // Timeout used when submitting data read from offline logs.
  private static final int LOG_TIMEOUT = 10;
  private static final TimeUnit LOG_TIMEOUT_UNIT = TimeUnit.SECONDS;
  private static final String COUNTERS_CATEGORY = "buck_offline_logs_stats";

  private final ScribeLogger scribeLogger;
  private final ImmutableList<String> blacklistCategories;
  private final int maxScribeOfflineLogsBytes;
  private final ProjectFilesystem filesystem;
  private final ObjectMapper objectMapper;

  @Nullable
  private BufferedOutputStream logFileStoreStream;
  private boolean storingAvailable;
  private int bytesStoredSoFar;
  private volatile boolean startedStoring;
  private volatile boolean startedClosing;
  private final Path logDir;
  private final Path newLogPath;
  private final AtomicBoolean startedSendingStored;
  private final IntegerCounter totalLinesResent;
  private final IntegerCounter totalBytesResent;
  private final IntegerCounter logfilesResent;

  public OfflineScribeLogger(
      ScribeLogger scribeLogger,
      ImmutableList<String> blacklistCategories,
      int maxScribeOfflineLogsKB,
      ProjectFilesystem projectFilesystem,
      ObjectMapper objectMapper,
      BuckEventBus buckEventBus,
      BuildId buildId) {
    Preconditions.checkNotNull(scribeLogger);
    Preconditions.checkNotNull(blacklistCategories);
    Preconditions.checkArgument(maxScribeOfflineLogsKB > 0);
    Preconditions.checkNotNull(projectFilesystem);
    Preconditions.checkNotNull(objectMapper);
    Preconditions.checkNotNull(buckEventBus);
    Preconditions.checkNotNull(buildId);

    this.scribeLogger = scribeLogger;
    this.blacklistCategories = blacklistCategories;
    this.maxScribeOfflineLogsBytes = KILO * maxScribeOfflineLogsKB;
    this.filesystem = projectFilesystem;
    this.objectMapper = objectMapper;

    this.logFileStoreStream = null;
    this.storingAvailable = true;
    this.bytesStoredSoFar = 0;
    this.startedStoring = false;
    this.startedClosing = false;
    this.logDir = projectFilesystem.getBuckPaths().getOfflineLogDir();
    this.newLogPath =
        projectFilesystem.resolve(logDir.resolve(LOGFILE_PREFIX + buildId + LOGFILE_SUFFIX));

    this.startedSendingStored = new AtomicBoolean(false);
    this.totalLinesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "total_lines_resent",
        ImmutableMap.of());
    this.totalBytesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "total_bytes_resent",
        ImmutableMap.of());
    this.logfilesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "logfiles_resent",
        ImmutableMap.of());
    buckEventBus.post(new CounterRegistry.AsyncCounterRegistrationEvent(
        ImmutableSet.of(totalLinesResent, totalBytesResent, logfilesResent)));
  }

  @Override
  public ListenableFuture<Void> log(final String category, final Iterable<String> lines) {
    ListenableFuture<Void> upload = scribeLogger.log(category, lines);
    Futures.addCallback(
        upload,
        new FutureCallback<Void>() {
          @Override
          public void onSuccess(Void result) {
            if (startedSendingStored.compareAndSet(false, true) && !startedStoring) {
              sendStoredLogs();
            }
          }

          @Override
          public void onFailure(Throwable t) {
            // Sending should be stopped if we require storing.
            startedStoring = true;

            if (!blacklistCategories.contains(category)) {
              LOG.verbose(t, "Storing Scribe lines from category: %s.", category);

              // Get data to store.
              byte[] scribeData;
              try {
                scribeData = objectMapper
                    .writeValueAsString(
                        ScribeData.builder()
                            .setCategory(category)
                            .setLines(lines)
                            .build())
                    .getBytes(Charsets.UTF_8);
              } catch (Exception e) {
                LOG.error(e, "Failed generating JSON to store for category: %s.", category);
                return;
              }

              storeOffline(category, scribeData);
            }
          }
        });

    return upload;
  }

  @Override
  public void close() throws IOException {
    // Request stopping sending as soon as possible.
    startedClosing = true;

    synchronized (this) {<commit_msg>Throw away exception rather than logging it<commit_after>/*
 * Copyright 2016-present Facebook, Inc.
 *
 * 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
 *
 *     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 com.facebook.buck.util.network.offline;

import com.facebook.buck.counters.CounterRegistry;
import com.facebook.buck.counters.IntegerCounter;
import com.facebook.buck.event.BuckEventBus;
import com.facebook.buck.io.ProjectFilesystem;
import com.facebook.buck.log.Logger;
import com.facebook.buck.model.BuildId;
import com.facebook.buck.util.network.ScribeLogger;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * This logger uses files-related operations (for offline logging i.e. storing and delayed logging
 * once network connection is better). Some of them may not be supported by all filesystems. E.g.
 * Jimfs uses JimfsPath which does not support toFile() operation. For such filesystems, the offline
 * logging will not be performed.
 */
public class OfflineScribeLogger extends ScribeLogger {

  protected static final String LOGFILE_PATTERN = "scribe-stored-*.log";
  protected static final String LOGFILE_PREFIX = "scribe-stored-";
  protected static final String LOGFILE_SUFFIX = ".log";

  private static final int BUFFER_SIZE = 1024 * 1024;
  private static final long CLUSTER_DISPATCH_SIZE = 1024 * 1024;
  private static final int KILO = 1024;
  private static final Logger LOG = Logger.get(OfflineScribeLogger.class);
  // Timeout used when submitting data read from offline logs.
  private static final int LOG_TIMEOUT = 10;
  private static final TimeUnit LOG_TIMEOUT_UNIT = TimeUnit.SECONDS;
  private static final String COUNTERS_CATEGORY = "buck_offline_logs_stats";

  private final ScribeLogger scribeLogger;
  private final ImmutableList<String> blacklistCategories;
  private final int maxScribeOfflineLogsBytes;
  private final ProjectFilesystem filesystem;
  private final ObjectMapper objectMapper;

  @Nullable
  private BufferedOutputStream logFileStoreStream;
  private boolean storingAvailable;
  private int bytesStoredSoFar;
  private volatile boolean startedStoring;
  private volatile boolean startedClosing;
  private final Path logDir;
  private final Path newLogPath;
  private final AtomicBoolean startedSendingStored;
  private final IntegerCounter totalLinesResent;
  private final IntegerCounter totalBytesResent;
  private final IntegerCounter logfilesResent;

  public OfflineScribeLogger(
      ScribeLogger scribeLogger,
      ImmutableList<String> blacklistCategories,
      int maxScribeOfflineLogsKB,
      ProjectFilesystem projectFilesystem,
      ObjectMapper objectMapper,
      BuckEventBus buckEventBus,
      BuildId buildId) {
    Preconditions.checkNotNull(scribeLogger);
    Preconditions.checkNotNull(blacklistCategories);
    Preconditions.checkArgument(maxScribeOfflineLogsKB > 0);
    Preconditions.checkNotNull(projectFilesystem);
    Preconditions.checkNotNull(objectMapper);
    Preconditions.checkNotNull(buckEventBus);
    Preconditions.checkNotNull(buildId);

    this.scribeLogger = scribeLogger;
    this.blacklistCategories = blacklistCategories;
    this.maxScribeOfflineLogsBytes = KILO * maxScribeOfflineLogsKB;
    this.filesystem = projectFilesystem;
    this.objectMapper = objectMapper;

    this.logFileStoreStream = null;
    this.storingAvailable = true;
    this.bytesStoredSoFar = 0;
    this.startedStoring = false;
    this.startedClosing = false;
    this.logDir = projectFilesystem.getBuckPaths().getOfflineLogDir();
    this.newLogPath =
        projectFilesystem.resolve(logDir.resolve(LOGFILE_PREFIX + buildId + LOGFILE_SUFFIX));

    this.startedSendingStored = new AtomicBoolean(false);
    this.totalLinesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "total_lines_resent",
        ImmutableMap.of());
    this.totalBytesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "total_bytes_resent",
        ImmutableMap.of());
    this.logfilesResent = new IntegerCounter(
        COUNTERS_CATEGORY,
        "logfiles_resent",
        ImmutableMap.of());
    buckEventBus.post(new CounterRegistry.AsyncCounterRegistrationEvent(
        ImmutableSet.of(totalLinesResent, totalBytesResent, logfilesResent)));
  }

  @Override
  public ListenableFuture<Void> log(final String category, final Iterable<String> lines) {
    ListenableFuture<Void> upload = scribeLogger.log(category, lines);
    Futures.addCallback(
        upload,
        new FutureCallback<Void>() {
          @Override
          public void onSuccess(Void result) {
            if (startedSendingStored.compareAndSet(false, true) && !startedStoring) {
              sendStoredLogs();
            }
          }

          @Override
          public void onFailure(Throwable t) {
            // Sending should be stopped if we require storing.
            startedStoring = true;

            if (!blacklistCategories.contains(category)) {
              LOG.verbose(t, "Storing Scribe lines from category: %s.", category);

              // Get data to store.
              byte[] scribeData;
              try {
                scribeData = objectMapper
                    .writeValueAsString(
                        ScribeData.builder()
                            .setCategory(category)
                            .setLines(lines)
                            .build())
                    .getBytes(Charsets.UTF_8);
              } catch (Exception e) {
                LOG.error(
                    "Failed generating JSON to store for category: %s: %s.",
                    category,
                    e.getMessage());
                return;
              }

              storeOffline(category, scribeData);
            }
          }
        });

    return upload;
  }

  @Override
  public void close() throws IOException {
    // Request stopping sending as soon as possible.
    startedClosing = true;

    synchronized (this) {
{'fraction_non_alphanumeric': 0.07826961770623743, 'fraction_numerical': 0.004426559356136821, 'mean_word_length': 3.416765402843602, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '533317', 'n_tokens_mistral': 4157, 'n_tokens_neox': 3978, 'n_words': 1056}