text: // Copyright 2022 PingCAP, Ltd.
  //
  // 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.
  
  #include <Common/Exception.h>
  #include <Flash/DiagnosticsService.h>
  #include <Flash/LogSearch.h>
  #include <Poco/DirectoryIterator.h>
  #include <Poco/Path.h>
  #include <Storages/Transaction/KVStore.h>
  #include <Storages/Transaction/ProxyFFI.h>
  #include <Storages/Transaction/TMTContext.h>
  #include <fmt/ranges.h>
  
  #include <ext/scope_guard.h>
  #include <memory>
  #include <vector>
  
  
  namespace DB
  {
  using diagnosticspb::LogLevel;
  using diagnosticspb::SearchLogResponse;
  
  ::grpc::Status DiagnosticsService::server_info(
      ::grpc::ServerContext *,
      const ::diagnosticspb::ServerInfoRequest * request,
      ::diagnosticspb::ServerInfoResponse * response)
  try
  {
      const TiFlashRaftProxyHelper * helper = server.context().getTMTContext().getKVStore()->getProxyHelper();
      if (helper)
      {
          std::string req = request->SerializeAsString();
          helper->fn_server_info(helper->proxy_ptr, strIntoView(&req), response);
      }
      else
      {
          LOG_FMT_ERROR(log, "TiFlashRaftProxyHelper is null, `DiagnosticsService::server_info` is useless");
          return ::grpc::Status(::grpc::StatusCode::INTERNAL, "TiFlashRaftProxyHelper is null");
      }
      return ::grpc::Status::OK;
  }
  catch (const Exception & e)
  {
      LOG_ERROR(log, e.displayText());
      return grpc::Status(grpc::StatusCode::INTERNAL, "internal error");
  }
  catch (const std::exception & e)
  {
      LOG_ERROR(log, e.what());
      return grpc::Status(grpc::StatusCode::INTERNAL, "internal error");
  }
  
  // get & filter(ts of last record < start-time) all files in same log directory.
  std::list<std::string> getFilesToSearch(IServer & server, Poco::Logger * log, const int64_t start_time)
  {
      std::list<std::string> files_to_search;
  
      std::string log_dir; // log directory
      auto error_log_file_prefix = server.config().getString("logger.errorlog", "*");
      auto tracing_log_file_prefix = server.config().getString("logger.tracing_log", "*");
      // ignore tiflash error log and mpp task tracing log
      std::vector<String> ignore_log_file_prefixes = {error_log_file_prefix, tracing_log_file_prefix};
  
      {
          auto log_file_prefix = server.config().getString("logger.log");
          if (auto it = log_file_prefix.rfind('/'); it != std::string::npos)
          {
              log_dir = std::string(log_file_prefix.begin(), log_file_prefix.begin() + it);
          }
      }
  
      LOG_FMT_DEBUG(log, "got log directory {}", log_dir);
  
      if (log_dir.empty())
          return files_to_search;
  
      Poco::DirectoryIterator dir_end;
      for (Poco::DirectoryIterator it(log_dir); it != dir_end; ++it)
      {
          const auto & path = it->path();
          if (it->isFile() && it->exists())
          {
              if (!FilterFileByDatetime(path, ignore_log_file_prefixes, start_time))
                  files_to_search.emplace_back(path);
          }
      }
  
      LOG_FMT_DEBUG(log, "got log files to search {}", files_to_search);
  
      return files_to_search;
  }
  
  grpc::Status searchLog(Poco::Logger * log, ::grpc::ServerWriter<::diagnosticspb::SearchLogResponse> * stream, LogIterator & log_itr)
  {
      static constexpr size_t LOG_BATCH_SIZE = 256;
  
      for (;;)
      {
          size_t i = 0;
          auto resp = SearchLogResponse::default_instance();
          for (; i < LOG_BATCH_SIZE;)
          {
              if (auto tmp_msg = log_itr.next(); !tmp_msg)
              {
                  break;
              }
              else
              {
                  i++;
                  resp.mutable_messages()->Add(std::move(*tmp_msg));
              }
          }
  
          if (i == 0)
              break;
  
          if (!stream->Write(resp))
          {
              LOG_FMT_DEBUG(log, "Write response failed for unknown reason.");
              return grpc::Status(grpc::StatusCode::UNKNOWN, "Write response failed for unknown reason.");
          }
      }
      return ::grpc::Status::OK;
  }
  
  ::grpc::Status DiagnosticsService::search_log(
      ::grpc::ServerContext *,
      const ::diagnosticspb::SearchLogRequest * request,
      ::grpc::ServerWriter<::diagnosticspb::SearchLogResponse> * stream)
  {
      int64_t start_time = request->start_time();
      int64_t end_time = request->end_time();
      if (end_time == 0)
      {
          end_time = std::numeric_limits<int64_t>::max();
      }
      std::vector<LogLevel> levels;
      for (auto level : request->levels())
      {
          levels.push_back(static_cast<LogLevel>(level));
      }
  
      std::vector<std::string> patterns;
      for (auto && pattern : request->patterns())
      {
          patterns.push_back(pattern);
      }
  
      LOG_FMT_DEBUG(log, "Handling SearchLog: {}", request->DebugString());
      SCOPE_EXIT({
          LOG_FMT_DEBUG(log, "Handling SearchLog done: {}", request->DebugString());
      });
  
      auto files_to_search = getFilesToSearch(server, log, start_time);
  
      for (const auto & path : files_to_search)
      {
          LOG_FMT_DEBUG(log, "start to search file {}", path);
          auto status = grpc::Status::OK;
          ReadLogFile(path, [&](std::istream & istr) {
              LogIterator log_itr(start_time, end_time, levels, patterns, istr);
              status = searchLog(log, stream, log_itr);
          });
          if (!status.ok())
              return status;
      }
  
      return ::grpc::Status::OK;
  }
  } // namespace DB
  