ceph/src/rgw/rgw_flight_frontend.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 /*
   5  * Ceph - scalable distributed file system
   6  *
   7  * Copyright 2023 IBM
   8  *
   9  * See file COPYING for licensing information.
  10  */
  11
  12 #pragma once
  13
  14 #include "include/common_fwd.h"
  15 #include "common/Thread.h"
  16 #include "rgw_frontend.h"
  17 #include "rgw_op.h"
  18
  19 #include "arrow/status.h"
  20
  21
  22 namespace rgw::flight {
  23
  24 using FlightKey = uint32_t;
  25 extern const FlightKey null_flight_key;
  26
  27 class FlightServer;
  28
  29 class FlightFrontend : public RGWFrontend {
  30
  31   static constexpr std::string_view server_thread_name =
  32     "Arrow Flight Server thread";
  33
  34   RGWProcessEnv& env;
  35   std::thread flight_thread;
  36   RGWFrontendConfig* config;
  37   int port;
  38
  39   const DoutPrefix dp;
  40
  41 public:
  42
  43   // port <= 0 means let server decide; typically 8077
  44   FlightFrontend(RGWProcessEnv& env,
  45                  RGWFrontendConfig* config,
  46                  int port = -1);
  47   ~FlightFrontend() override;
  48   int init() override;
  49   int run() override;
  50   void stop() override;
  51   void join() override;
  52
  53   void pause_for_new_config() override;
  54   void unpause_with_new_config() override;
  55 }; // class FlightFrontend
  56
  57 class FlightGetObj_Filter : public RGWGetObj_Filter {
  58
  59   const RGWProcessEnv& penv;
  60   const DoutPrefix dp;
  61   FlightKey key;
  62   uint64_t current_offset;
  63   uint64_t expected_size;
  64   std::string uri;
  65   std::string tenant_name;
  66   std::string bucket_name;
  67   rgw_obj_key object_key;
  68   std::string temp_file_name;
  69   std::ofstream temp_file;
  70   arrow::Status schema_status;
  71   rgw_user user_id; // TODO: this should be removed when we do
  72   // proper flight authentication
  73
  74 public:
  75
  76   FlightGetObj_Filter(const req_state* request, RGWGetObj_Filter* next);
  77   ~FlightGetObj_Filter();
  78
  79   int handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len) override;
  80 #if 0
  81   // this would allow the range to be modified if necessary;
  82   int fixup_range(off_t& ofs, off_t& end) override;
  83 #endif
  84 };
  85
  86 } // namespace rgw::flight