PostgreSQL.hpp 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. /*
  2. * Copyright (c)2019 ZeroTier, Inc.
  3. *
  4. * Use of this software is governed by the Business Source License included
  5. * in the LICENSE.TXT file in the project's root directory.
  6. *
  7. * Change Date: 2025-01-01
  8. *
  9. * On the date above, in accordance with the Business Source License, use
  10. * of this software will be governed by version 2.0 of the Apache License.
  11. */
  12. /****/
  13. #include "DB.hpp"
  14. #ifdef ZT_CONTROLLER_USE_LIBPQ
  15. #ifndef ZT_CONTROLLER_LIBPQ_HPP
  16. #define ZT_CONTROLLER_LIBPQ_HPP
  17. #define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4
  18. #include <memory>
  19. #include <redis++/redis++.h>
  20. extern "C" {
  21. typedef struct pg_conn PGconn;
  22. }
  23. namespace ZeroTier {
  24. struct RedisConfig;
  25. /**
  26. * A controller database driver that talks to PostgreSQL
  27. *
  28. * This is for use with ZeroTier Central. Others are free to build and use it
  29. * but be aware taht we might change it at any time.
  30. */
  31. class PostgreSQL : public DB
  32. {
  33. public:
  34. PostgreSQL(const Identity &myId, const char *path, int listenPort, RedisConfig *rc);
  35. virtual ~PostgreSQL();
  36. virtual bool waitForReady();
  37. virtual bool isReady();
  38. virtual bool save(nlohmann::json &record,bool notifyListeners);
  39. virtual void eraseNetwork(const uint64_t networkId);
  40. virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
  41. virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress &physicalAddress);
  42. virtual void updateMemberOnLoad(const uint64_t networkId, const uint64_t memberId, nlohmann::json &member);
  43. protected:
  44. struct _PairHasher
  45. {
  46. inline std::size_t operator()(const std::pair<uint64_t,uint64_t> &p) const { return (std::size_t)(p.first ^ p.second); }
  47. };
  48. private:
  49. void initializeNetworks(PGconn *conn);
  50. void initializeMembers(PGconn *conn);
  51. void heartbeat();
  52. void membersDbWatcher();
  53. void _membersWatcher_Postgres(PGconn *conn);
  54. void networksDbWatcher();
  55. void _networksWatcher_Postgres(PGconn *conn);
  56. void _membersWatcher_Redis();
  57. void _networksWatcher_Redis();
  58. void commitThread();
  59. void onlineNotificationThread();
  60. void onlineNotification_Postgres();
  61. void onlineNotification_Redis();
  62. void _doRedisUpdate(sw::redis::Transaction &tx, std::string &controllerId,
  63. std::unordered_map< std::pair<uint64_t,uint64_t>,std::pair<int64_t,InetAddress>,_PairHasher > &lastOnline);
  64. enum OverrideMode {
  65. ALLOW_PGBOUNCER_OVERRIDE = 0,
  66. NO_OVERRIDE = 1
  67. };
  68. PGconn * getPgConn( OverrideMode m = ALLOW_PGBOUNCER_OVERRIDE );
  69. const Identity _myId;
  70. const Address _myAddress;
  71. std::string _myAddressStr;
  72. std::string _connString;
  73. BlockingQueue< std::pair<nlohmann::json,bool> > _commitQueue;
  74. std::thread _heartbeatThread;
  75. std::thread _membersDbWatcher;
  76. std::thread _networksDbWatcher;
  77. std::thread _commitThread[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS];
  78. std::thread _onlineNotificationThread;
  79. std::unordered_map< std::pair<uint64_t,uint64_t>,std::pair<int64_t,InetAddress>,_PairHasher > _lastOnline;
  80. mutable std::mutex _lastOnline_l;
  81. mutable std::mutex _readyLock;
  82. std::atomic<int> _ready, _connected, _run;
  83. mutable volatile bool _waitNoticePrinted;
  84. int _listenPort;
  85. uint8_t _ssoPsk[48];
  86. RedisConfig *_rc;
  87. std::shared_ptr<sw::redis::Redis> _redis;
  88. std::shared_ptr<sw::redis::RedisCluster> _cluster;
  89. };
  90. } // namespace ZeroTier
  91. #endif // ZT_CONTROLLER_LIBPQ_HPP
  92. #endif // ZT_CONTROLLER_USE_LIBPQ