text: // 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.
  #pragma once
  
  #include "kudu/consensus/log.h"
  
  #include <glog/logging.h>
  #include <gtest/gtest.h>
  
  #include <memory>
  #include <string>
  #include <utility>
  #include <vector>
  
  #include "kudu/clock/clock.h"
  #include "kudu/clock/hybrid_clock.h"
  #include "kudu/common/timestamp.h"
  #include "kudu/common/wire_protocol-test-util.h"
  #include "kudu/consensus/log_anchor_registry.h"
  #include "kudu/consensus/log_reader.h"
  #include "kudu/consensus/log_util.h"
  #include "kudu/consensus/metadata.pb.h"
  #include "kudu/consensus/opid_util.h"
  #include "kudu/fs/fs_manager.h"
  #include "kudu/gutil/stl_util.h"
  #include "kudu/gutil/stringprintf.h"
  #include "kudu/gutil/strings/substitute.h"
  #include "kudu/gutil/strings/util.h"
  #include "kudu/tserver/tserver.pb.h"
  #include "kudu/util/async_util.h"
  #include "kudu/util/env_util.h"
  #include "kudu/util/file_cache.h"
  #include "kudu/util/metrics.h"
  #include "kudu/util/path_util.h"
  #include "kudu/util/pb_util.h"
  #include "kudu/util/stopwatch.h"
  #include "kudu/util/test_macros.h"
  #include "kudu/util/test_util.h"
  
  METRIC_DECLARE_entity(server);
  METRIC_DECLARE_entity(tablet);
  
  namespace kudu {
  namespace log {
  
  constexpr char kTestTable[] = "test-log-table";
  constexpr char kTestTableId[] = "test-log-table-id";
  constexpr char kTestTablet[] = "test-log-tablet";
  constexpr bool APPEND_SYNC = true;
  constexpr bool APPEND_ASYNC = false;
  
  // Append a single batch of 'count' NoOps to the log.
  // If 'size' is not NULL, increments it by the expected increase in log size.
  // Increments 'op_id''s index once for each operation logged.
  inline Status AppendNoOpsToLogSync(clock::Clock* clock,
                                     Log* log,
                                     consensus::OpId* op_id,
                                     int count,
                                     int* size = nullptr) {
  
    std::vector<consensus::ReplicateRefPtr> replicates;
    for (int i = 0; i < count; i++) {
      consensus::ReplicateRefPtr replicate =
          make_scoped_refptr_replicate(new consensus::ReplicateMsg());
      consensus::ReplicateMsg* repl = replicate->get();
  
      repl->mutable_id()->CopyFrom(*op_id);
      repl->set_op_type(consensus::NO_OP);
      repl->set_timestamp(clock->Now().ToUint64());
  
      // Increment op_id.
      op_id->set_index(op_id->index() + 1);
  
      if (size) {
        // If we're tracking the sizes we need to account for the fact that the Log wraps the
        // log entry in an LogEntryBatchPB, and each actual entry will have a one-byte tag.
        *size += repl->ByteSize() + 1;
      }
      replicates.push_back(replicate);
    }
  
    // Account for the entry batch header and wrapper PB.
    if (size) {
      *size += log::kEntryHeaderSizeV2 + 5;
    }
  
    Synchronizer s;
    RETURN_NOT_OK(log->AsyncAppendReplicates(replicates,
                                             s.AsStatusCallback()));
    return s.Wait();
  }
  
  inline Status AppendNoOpToLogSync(clock::Clock* clock,
                                    Log* log,
                                    consensus::OpId* op_id,
                                    int* size = nullptr) {
    return AppendNoOpsToLogSync(clock, log, op_id, 1, size);
  }
  
  
  // Corrupts the last segment of the provided log by either truncating it
  // or modifying a byte at the given offset.
  enum CorruptionType {
    TRUNCATE_FILE,
    FLIP_BYTE
  };
  
  inline Status CorruptLogFile(Env* env, const std::string& log_path,
                               CorruptionType type, int corruption_offset) {
    faststring buf;
    RETURN_NOT_OK_PREPEND(ReadFileToString(env, log_path, &buf),
                          "Couldn't read log");
  
    switch (type) {
      case TRUNCATE_FILE:
        buf.resize(corruption_offset);
        break;
      case FLIP_BYTE:
        CHECK_LT(corruption_offset, buf.size());
        buf[corruption_offset] ^= 0xff;
        break;
    }
  
    // Rewrite the file with the corrupt log.
    RETURN_NOT_OK_PREPEND(WriteStringToFile(env, Slice(buf), log_path),
                          "Couldn't rewrite corrupt log file");
  
    return Status::OK();
  }
  
  class LogTestBase : public KuduTest {
   public:
    typedef std::pair<int, int> DeltaId;
  
    LogTestBase()
        : schema_(GetSimpleTestSchema()),
          log_anchor_registry_(new LogAnchorRegistry) {
    }
  
    void SetUp() override {
      KuduTest::SetUp();
      current_index_ = kStartIndex;
      fs_manager_.reset(new FsManager(env_, FsManagerOpts(GetTestPath("fs_root"))));
      metric_registry_.reset(new MetricRegistry);
      metric_entity_tablet_ = METRIC_ENTITY_tablet.Instantiate(
          metric_registry_.get(), "tablet");
      metric_entity_server_ = METRIC_ENTITY_server.Instantiate(
          metric_registry_.get(), "server");
      // Capacity was chosen arbitrarily: high enough to cache multiple files, but
      // low enough to see some eviction.
      file_cache_.reset(new FileCache("log-test-base", env_, 5, metric_entity_server_));
      ASSERT_OK(file_cache_->Init());
      ASSERT_OK(fs_manager_->CreateInitialFileSystemLayout());
      ASSERT_OK(fs_manager_->Open());
  
      clock_.reset(new clock::HybridClock(metric_entity_server_));
      ASSERT_OK(clock_->Init());
    }
  
    void TearDown() override {
      KuduTest::TearDown();
    }
  
    Status BuildLog() {
      Schema schema_with_ids = SchemaBuilder(schema_).Build();
      return Log::Open(options_,
                       fs_manager_.get(),
                       file_cache_.get(),
                       kTestTablet,
                       schema_with_ids,
                       0, // schema_version
                       metric_entity_tablet_.get(),
                       &log_);
    }
  
    void CheckRightNumberOfSegmentFiles(int expected) {
      // Test that we actually have the expected number of files in the fs.
      // We should have n segments plus '.' and '..'
      std::vector<std::string> files;
      ASSERT_OK(env_->GetChildren(
                         JoinPathSegments(fs_manager_->GetWalsRootDir(),
                                          kTestTablet),
                         &files));
      int count = 0;
      for (const std::string& s : files) {
        if (HasPrefixString(s, FsManager::kWalFileNamePrefix)) {
          count++;
        }
      }
      ASSERT_EQ(expected, count);
    }
  
    void EntriesToIdList(std::vector<uint32_t>* ids) {
      for (const auto& entry : entries_) {
        VLOG(2) << "Entry contents: " << pb_util::SecureDebugString(*entry);
        if (entry->type() == REPLICATE) {
          ids->push_back(entry->replicate().id().index());
        }
      }
    }
  
    static void CheckReplicateResult(const consensus::ReplicateRefPtr& msg, const Status& s) {
      CHECK_OK(s);
    }
  
    // Appends a batch with size 2 (1 insert, 1 mutate) to the log.
    Status AppendReplicateBatch(const consensus::OpId& opid,
                                bool sync = APPEND_SYNC) {
      consensus::ReplicateRefPtr replicate =
          make_scoped_refptr_replicate(new consensus::ReplicateMsg());
      replicate->get()->set_op_type(consensus::WRITE_OP);
      replicate->get()->mutable_id()->CopyFrom(opid);
      replicate->get()->set_timestamp(clock_->Now().ToUint64());
      tserver::WriteRequestPB* batch_request = replicate->get()->mutable_write_request();
      RETURN_NOT_OK(SchemaToPB(schema_, batch_request->mutable_schema()));
      AddTestRowToPB(RowOperationsPB::INSERT, schema_,
                     opid.index(),
                     0,
                     "this is a test insert",
                     batch_request->mutable_row_operations());
      AddTestRowToPB(RowOperationsPB::UPDATE, schema_,
                     opid.index() + 1,
                     0,
                     "this is a test mutate",
                     batch_request->mutable_row_operations());
      batch_request->set_tablet_id(kTestTablet);
      return AppendReplicateBatch(replicate, sync);
    }
  
    // Appends the provided batch to the log.
    Status AppendReplicateBatch(const consensus::ReplicateRefPtr& replicate,
                                bool sync = APPEND_SYNC) {
      if (sync) {
        Synchronizer s;
        RETURN_NOT_OK(log_->AsyncAppendReplicates({ replicate }, s.AsStatusCallback()));
        return s.Wait();
      }
      // AsyncAppendReplicates does not free the ReplicateMsg on completion, so we
      // need to pass it through to our callback.
      return log_->AsyncAppendReplicates(
          { replicate }, Bind(&LogTestBase::CheckReplicateResult, replicate));
    }
  
    static void CheckCommitResult(const Status& s) {
      CHECK_OK(s);
    }
  
    // Append a commit log entry containing one entry for the insert and one
    // for the mutate.
    Status AppendCommit(const consensus::OpId& original_opid,
                        bool sync = APPEND_SYNC) {
      // The mrs id for the insert.
      constexpr int kTargetMrsId = 1;
  
      // The rs and delta ids for the mutate.
      constexpr int kTargetRsId = 0;
      constexpr int kTargetDeltaId = 0;
  
      return AppendCommit(original_opid, kTargetMrsId, kTargetRsId, kTargetDeltaId, sync);
    }
  
    Status AppendCommit(const consensus::OpId& original_opid,
                        int mrs_id,
                        int rs_id,
                        int dms_id,
                        bool sync = APPEND_SYNC) {
      std::unique_ptr<consensus::CommitMsg> commit(new consensus::CommitMsg);
      commit->set_op_type(consensus::WRITE_OP);
  
      commit->mutable_commited_op_id()->CopyFrom(original_opid);
  
      tablet::TxResultPB* result = commit->mutable_result();
  
      tablet::OperationResultPB* insert = result->add_ops();
      insert->add_mutated_stores()->set_mrs_id(mrs_id);
  
      tablet::OperationResultPB* mutate = result->add_ops();
      tablet::MemStoreTargetPB* target = mutate->add_mutated_stores();
      target->set_dms_id(dms_id);
      target->set_rs_id(rs_id);
      return AppendCommit(std::move(commit), sync);
    }
  
    // Append a COMMIT message for 'original_opid', but with results
    // indicating that the associated writes failed due to
    // "NotFound" errors.
    Status AppendCommitWithNotFoundOpResults(const consensus::OpId& original_opid) {
      std::unique_ptr<consensus::CommitMsg> commit(new consensus::CommitMsg);
      commit->set_op_type(consensus::WRITE_OP);
      commit->mutable_commited_op_id()->CopyFrom(original_opid);
  
      tablet::TxResultPB* result = commit->mutable_result();
  
      tablet::OperationResultPB* insert = result->add_ops();
      StatusToPB(Status::NotFound("fake failed write"), insert->mutable_failed_status());
      tablet::OperationResultPB* mutate = result->add_ops();
      StatusToPB(Status::NotFound("fake failed write"), mutate->mutable_failed_status());
  
      return AppendCommit(std::move(commit));
    }
  
    Status AppendCommit(std::unique_ptr<consensus::CommitMsg> commit,
                        bool sync = APPEND_SYNC) {
      if (sync) {
        Synchronizer s;
        RETURN_NOT_OK(log_->AsyncAppendCommit(std::move(commit), s.AsStatusCallback()));
        return s.Wait();
      }
      return log_->AsyncAppendCommit(std::move(commit),
                                     Bind(&LogTestBase::CheckCommitResult));
    }
  
      // Appends 'count' ReplicateMsgs and the corresponding CommitMsgs to the log
    Status AppendReplicateBatchAndCommitEntryPairsToLog(int count,
                                                        bool sync = APPEND_SYNC) {
      for (int i = 0; i < count; i++) {
        consensus::OpId opid = consensus::MakeOpId(1, current_index_);
        RETURN_NOT_OK(AppendReplicateBatch(opid));
        RETURN_NOT_OK(AppendCommit(opid, sync));
        current_index_ += 1;
      }
      return Status::OK();
    }
  
    // Append a single NO_OP entry. Increments op_id by one.
    // If non-NULL, and if the write is successful, 'size' is incremented
    // by the size of the written operation.
    Status AppendNoOp(consensus::OpId* op_id, int* size = nullptr) {
      return AppendNoOpToLogSync(clock_.get(), log_.get(), op_id, size);
    }
  
    // Append a number of no-op entries to the log.
    // Increments op_id's index by the number of records written.
    // If non-NULL, 'size' keeps track of the size of the operations
    // successfully written.
    Status AppendNoOps(consensus::OpId* op_id, int num, int* size = nullptr) {
      for (int i = 0; i < num; i++) {
        RETURN_NOT_OK(AppendNoOp(op_id, size));
      }
      return Status::OK();
    }
  
    Status RollLog() {
      return log_->AllocateSegmentAndRollOverForTests();
    }
  
    static std::string DumpSegmentsToString(const SegmentSequence& segments) {
      std::string dump;
      for (const scoped_refptr<ReadableLogSegment>& segment : segments) {
        dump.append("------------\n");
        strings::SubstituteAndAppend(&dump, "Segment: $0, Path: $1\n",
                                     segment->header().sequence_number(), segment->path());
        strings::SubstituteAndAppend(&dump, "Header: $0\n",
                                     pb_util::SecureShortDebugString(segment->header()));
        if (segment->HasFooter()) {
          strings::SubstituteAndAppend(&dump, "Footer: $0\n",
                                       pb_util::SecureShortDebugString(segment->footer()));
        } else {
          dump.append("Footer: None or corrupt.");
        }
      }
      return dump;
    }
  
   protected:
    enum {
      kStartIndex = 1
    };
  
    const Schema schema_;
    std::unique_ptr<FsManager> fs_manager_;
    std::unique_ptr<MetricRegistry> metric_registry_;
    std::unique_ptr<FileCache> file_cache_;
    scoped_refptr<MetricEntity> metric_entity_tablet_;
    scoped_refptr<MetricEntity> metric_entity_server_;
    scoped_refptr<Log> log_;
    int64_t current_index_;
    LogOptions options_;
    // Reusable entries vector that deletes the entries on destruction.
    LogEntries entries_;
    scoped_refptr<LogAnchorRegistry> log_anchor_registry_;
    std::unique_ptr<clock::Clock> clock_;
  };
  
  } // namespace log
  } // namespace kudu
  