|
| 1 | +#include "contrib/reverse_tunnel_reporter/source/reporters/event_reporter/reporter.h" |
| 2 | + |
| 3 | +#include "source/common/protobuf/utility.h" |
| 4 | + |
| 5 | +namespace Envoy { |
| 6 | +namespace Extensions { |
| 7 | +namespace Bootstrap { |
| 8 | +namespace ReverseConnection { |
| 9 | + |
| 10 | +EventReporter::EventReporter(Server::Configuration::ServerFactoryContext& context, |
| 11 | + const ConfigProto& config, |
| 12 | + std::vector<ReverseTunnelReporterClientPtr>&& clients) |
| 13 | + : context_{context}, clients_{std::move(clients)}, |
| 14 | + stats_(generateStats( |
| 15 | + PROTOBUF_GET_STRING_OR_DEFAULT(config, stat_prefix, "reverse_tunnel_reporter"), |
| 16 | + context.scope())) { |
| 17 | + ENVOY_LOG(info, "EventReporter: Constructed with {} clients", clients_.size()); |
| 18 | +} |
| 19 | + |
| 20 | +void EventReporter::onServerInitialized() { |
| 21 | + ENVOY_LOG(info, "EventReporter: Initialized"); |
| 22 | + for (auto& client : clients_) { |
| 23 | + client->onServerInitialized(this); |
| 24 | + } |
| 25 | +} |
| 26 | + |
| 27 | +void EventReporter::reportConnectionEvent(absl::string_view node_id, absl::string_view cluster_id, |
| 28 | + absl::string_view tenant_id) { |
| 29 | + auto ptr = std::make_shared<ReverseTunnelEvent::Connected>( |
| 30 | + ReverseTunnelEvent::Connected{std::string(node_id), std::string(cluster_id), |
| 31 | + std::string(tenant_id), Envoy::SystemTime::clock::now()}); |
| 32 | + |
| 33 | + context_.mainThreadDispatcher().post( |
| 34 | + [this, ptr = std::move(ptr)]() mutable { this->addConnection(std::move(ptr)); }); |
| 35 | +} |
| 36 | + |
| 37 | +void EventReporter::reportDisconnectionEvent(absl::string_view node_id, |
| 38 | + absl::string_view cluster_id) { |
| 39 | + std::string name = ReverseTunnelEvent::getName(node_id, cluster_id); |
| 40 | + auto ptr = std::make_shared<ReverseTunnelEvent::Disconnected>( |
| 41 | + ReverseTunnelEvent::Disconnected{std::move(name)}); |
| 42 | + |
| 43 | + context_.mainThreadDispatcher().post( |
| 44 | + [this, ptr = std::move(ptr)]() mutable { this->removeConnection(std::move(ptr)); }); |
| 45 | +} |
| 46 | + |
| 47 | +// This is only served on the main thread so no locks needed. |
| 48 | +ReverseTunnelEvent::SharedConnections EventReporter::getAllConnections() { |
| 49 | + ASSERT(context_.mainThreadDispatcher().isThreadSafe()); |
| 50 | + stats_.reverse_tunnel_full_pulls_total_.inc(); |
| 51 | + |
| 52 | + ReverseTunnelEvent::SharedConnections all_connections; |
| 53 | + all_connections.reserve(connections_.size()); |
| 54 | + |
| 55 | + for (auto& [key, val] : connections_) { |
| 56 | + all_connections.push_back(val.connection); |
| 57 | + } |
| 58 | + return all_connections; |
| 59 | +} |
| 60 | + |
| 61 | +EventReporterStats EventReporter::generateStats(const std::string& prefix, Stats::Scope& scope) { |
| 62 | + return EventReporterStats{ALL_EVENT_REPORTER_STATS(POOL_COUNTER_PREFIX(scope, prefix), |
| 63 | + POOL_GAUGE_PREFIX(scope, prefix))}; |
| 64 | +} |
| 65 | + |
| 66 | +void EventReporter::notifyClients(ReverseTunnelEvent::BatchedEvents&& batch) { |
| 67 | + for (auto& client : clients_) { |
| 68 | + client->receiveEvents(batch); |
| 69 | + } |
| 70 | +} |
| 71 | + |
| 72 | +void EventReporter::addConnection(std::shared_ptr<ReverseTunnelEvent::Connected>&& connection) { |
| 73 | + ASSERT(context_.mainThreadDispatcher().isThreadSafe()); |
| 74 | + |
| 75 | + ENVOY_LOG(info, "EventReporter: Accepted a new connection. Node: {}, Cluster: {}, Tenant: {}", |
| 76 | + connection->node_id, connection->cluster_id, connection->tenant_id); |
| 77 | + |
| 78 | + std::string name = ReverseTunnelEvent::getName(connection->node_id, connection->cluster_id); |
| 79 | + auto [it, inserted] = |
| 80 | + connections_.try_emplace(std::move(name), ConnectionEntry{std::move(connection), 1}); |
| 81 | + |
| 82 | + if (inserted) { |
| 83 | + stats_.reverse_tunnel_unique_active_.inc(); |
| 84 | + notifyClients(ReverseTunnelEvent::BatchedEvents{{it->second.connection}, {}}); |
| 85 | + } else { |
| 86 | + // Multiple reverse tunnels can share the same name (same node). |
| 87 | + // We ref-count them and only notify clients of removal when the last one disconnects. |
| 88 | + it->second.count++; |
| 89 | + } |
| 90 | + |
| 91 | + stats_.reverse_tunnel_established_total_.inc(); |
| 92 | + stats_.reverse_tunnel_active_.inc(); |
| 93 | +} |
| 94 | + |
| 95 | +void EventReporter::removeConnection( |
| 96 | + std::shared_ptr<ReverseTunnelEvent::Disconnected>&& disconnection) { |
| 97 | + ASSERT(context_.mainThreadDispatcher().isThreadSafe()); |
| 98 | + |
| 99 | + const auto& name = disconnection->name; |
| 100 | + auto it = connections_.find(name); |
| 101 | + |
| 102 | + ENVOY_LOG(info, "EventReporter: Removed connection. Name: {}", name); |
| 103 | + |
| 104 | + if (it == connections_.end()) { |
| 105 | + ENVOY_LOG(warn, "EventReporter: Tried to remove a connection which doesnt exist"); |
| 106 | + return; |
| 107 | + } |
| 108 | + |
| 109 | + // Only notify removal on the last ref — see addConnection for the ref-count rationale. |
| 110 | + if (it->second.count == 1) { |
| 111 | + connections_.erase(it); |
| 112 | + stats_.reverse_tunnel_unique_active_.dec(); |
| 113 | + notifyClients(ReverseTunnelEvent::BatchedEvents{{}, {disconnection}}); |
| 114 | + } else { |
| 115 | + it->second.count--; |
| 116 | + } |
| 117 | + |
| 118 | + stats_.reverse_tunnel_closed_total_.inc(); |
| 119 | + stats_.reverse_tunnel_active_.dec(); |
| 120 | +} |
| 121 | + |
| 122 | +} // namespace ReverseConnection |
| 123 | +} // namespace Bootstrap |
| 124 | +} // namespace Extensions |
| 125 | +} // namespace Envoy |
0 commit comments