<commit_before>// 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.

#include <unistd.h>

#include <new>

#include <functional>
#include <string>

#include <process/dispatch.hpp>
#include <process/io.hpp>
#include <process/process.hpp>

#include <stout/bytes.hpp>
#include <stout/error.hpp>
#include <stout/exit.hpp>
#include <stout/nothing.hpp>
#include <stout/path.hpp>
#include <stout/stringify.hpp>
#include <stout/try.hpp>

#include <stout/os/shell.hpp>
#include <stout/os/write.hpp>

#include "slave/container_loggers/logrotate.hpp"


using namespace process;
using namespace mesos::internal::logger::rotate;


class LogrotateLoggerProcess : public Process<LogrotateLoggerProcess>
{
public:
  LogrotateLoggerProcess(const Flags& _flags)
    : flags(_flags),
      leading(None()),
      bytesWritten(0)
  {
    // Prepare a buffer for reading from the `incoming` pipe.
    length = sysconf(_SC_PAGE_SIZE);
    buffer = new char[length];
  }

  virtual ~LogrotateLoggerProcess()
  {
    if (buffer != nullptr) {
      delete[] buffer;
      buffer = nullptr;
    }

    if (leading.isSome()) {
      os::close(leading.get());
    }
  }

  // Prepares and starts the loop which reads from stdin, writes to the
  // leading log file, and manages total log size.
  Future<Nothing> run()
  {
    // Populate the `logrotate` configuration file.
    // See `Flags::logrotate_options` for the format.
    //
    // NOTE: We specify a size of `--max_size - length` because `logrotate`
    // has slightly different size semantics.  `logrotate` will rotate when the
    // max size is *exceeded*.  We rotate to keep files *under* the max size.
    const std::string config =
      "\"" + flags.log_filename.get() + "\" {\n" +
      flags.logrotate_options.getOrElse("") + "\n" +
      "size " + stringify(flags.max_size.bytes() - length) + "\n" +
      "}";

    Try<Nothing> result = os::write(
        flags.log_filename.get() + CONF_SUFFIX, config);

    if (result.isError()) {
      return Failure("Failed to write configuration file: " + result.error());
    }

    // NOTE: This is a prerequisuite for `io::read`.
    Try<Nothing> nonblock = os::nonblock(STDIN_FILENO);
    if (nonblock.isError()) {
      return Failure("Failed to set nonblocking pipe: " + nonblock.error());
    }

    // NOTE: This does not block.
    loop();

    return promise.future();
  }

  // Reads from stdin and writes to the leading log file.
  void loop()
  {
    io::read(STDIN_FILENO, buffer, length)
      .then([&](size_t readSize) -> Future<Nothing> {
        // Check if EOF has been reached on the input stream.
        // This indicates that the container (whose logs are being
        // piped to this process) has exited.
        if (readSize <= 0) {
          promise.set(Nothing());
          return Nothing();
        }

        // Do log rotation (if necessary) and write the bytes to the
        // leading log file.
        Try<Nothing> result = write(readSize);
        if (result.isError()) {
          promise.fail("Failed to write: " + result.error());
          return Nothing();
        }

        // Use `dispatch` to limit the size of the call stack.
        dispatch(self(), &LogrotateLoggerProcess::loop);

        return Nothing();
      });
  }

  // Writes the buffer from stdin to the leading log file.
  // When the number of written bytes exceeds `--max_size`, the leading<commit_msg>Fixed continuation logic in logrotate module.<commit_after>// 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.

#include <unistd.h>

#include <new>

#include <functional>
#include <string>

#include <process/defer.hpp>
#include <process/dispatch.hpp>
#include <process/io.hpp>
#include <process/process.hpp>

#include <stout/bytes.hpp>
#include <stout/error.hpp>
#include <stout/exit.hpp>
#include <stout/nothing.hpp>
#include <stout/path.hpp>
#include <stout/stringify.hpp>
#include <stout/try.hpp>

#include <stout/os/shell.hpp>
#include <stout/os/write.hpp>

#include "slave/container_loggers/logrotate.hpp"


using namespace process;
using namespace mesos::internal::logger::rotate;


class LogrotateLoggerProcess : public Process<LogrotateLoggerProcess>
{
public:
  LogrotateLoggerProcess(const Flags& _flags)
    : flags(_flags),
      leading(None()),
      bytesWritten(0)
  {
    // Prepare a buffer for reading from the `incoming` pipe.
    length = sysconf(_SC_PAGE_SIZE);
    buffer = new char[length];
  }

  virtual ~LogrotateLoggerProcess()
  {
    if (buffer != nullptr) {
      delete[] buffer;
      buffer = nullptr;
    }

    if (leading.isSome()) {
      os::close(leading.get());
    }
  }

  // Prepares and starts the loop which reads from stdin, writes to the
  // leading log file, and manages total log size.
  Future<Nothing> run()
  {
    // Populate the `logrotate` configuration file.
    // See `Flags::logrotate_options` for the format.
    //
    // NOTE: We specify a size of `--max_size - length` because `logrotate`
    // has slightly different size semantics.  `logrotate` will rotate when the
    // max size is *exceeded*.  We rotate to keep files *under* the max size.
    const std::string config =
      "\"" + flags.log_filename.get() + "\" {\n" +
      flags.logrotate_options.getOrElse("") + "\n" +
      "size " + stringify(flags.max_size.bytes() - length) + "\n" +
      "}";

    Try<Nothing> result = os::write(
        flags.log_filename.get() + CONF_SUFFIX, config);

    if (result.isError()) {
      return Failure("Failed to write configuration file: " + result.error());
    }

    // NOTE: This is a prerequisuite for `io::read`.
    Try<Nothing> nonblock = os::nonblock(STDIN_FILENO);
    if (nonblock.isError()) {
      return Failure("Failed to set nonblocking pipe: " + nonblock.error());
    }

    // NOTE: This does not block.
    loop();

    return promise.future();
  }

  // Reads from stdin and writes to the leading log file.
  void loop()
  {
    io::read(STDIN_FILENO, buffer, length)
      .then(defer(self(), [&](size_t readSize) -> Future<Nothing> {
        // Check if EOF has been reached on the input stream.
        // This indicates that the container (whose logs are being
        // piped to this process) has exited.
        if (readSize <= 0) {
          promise.set(Nothing());
          return Nothing();
        }

        // Do log rotation (if necessary) and write the bytes to the
        // leading log file.
        Try<Nothing> result = write(readSize);
        if (result.isError()) {
          promise.fail("Failed to write: " + result.error());
          return Nothing();
        }

        // Use `dispatch` to limit the size of the call stack.
        dispatch(self(), &LogrotateLoggerProcess::loop);

        return Nothing();
      }));
  }

  // Writes the buffer from stdin to the leading log file.
  // When the number of written bytes exceeds `--max_size`, the leading
{'fraction_non_alphanumeric': 0.12177342256214149, 'fraction_numerical': 0.0014340344168260039, 'mean_word_length': 2.8998136067101585, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 50, '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': '6259380', 'n_tokens_mistral': 2483, 'n_tokens_neox': 2408, 'n_words': 839}