Add gRPC address to WorkerInterface

Author: vishesh

fdbcli_lib/CliCommands.cpp

@@ -274,6 +274,7 @@ Future<grpc::Status> getWorkers(Reference<IDatabase> db, const GetWorkersRequest
 					Worker* w = rep->add_workers();
 					w->set_address(data.address.toString());
 					w->set_process_class(data.processClass.toString());
+					w->set_grpc_address(w->grpc_address());
 
 					auto* lc = w->mutable_locality();
 					localityDataToProto(data.locality, lc);

fdbcli_lib/ExcludeCommand.cpp

@@ -345,7 +345,7 @@ Future<grpc::Status> exclude(Reference<IDatabase> db, const ExcludeRequest* req,
 			else if (addr.port > 0 && worker->second.count(addr.port) == 0)
 				absentExclusions.insert(addr);
 		}
-		
+
 		co_return grpc::Status::OK;
 	} catch (const Error& e) {
 		co_return grpc::Status(grpc::StatusCode::INTERNAL,

fdbcli_lib/include/fdbcli_lib/CliCommands.h

@@ -73,7 +73,7 @@ Future<std::set<NetworkAddress>> getInProgressExclusion(Reference<ITransaction>
 
 Future<Void> getStorageServerInterfaces(Reference<IDatabase> db,
                                         std::map<std::string, StorageServerInterface>* interfaces);
-Future<bool> getWorkers(Reference<IDatabase> db, std::vector<ProcessData>* workers);    
+Future<bool> getWorkers(Reference<IDatabase> db, std::vector<ProcessData>* workers);
 } // namespace utils
 
 namespace special_keys {

fdbclient/CoordinationInterface.cpp

@@ -37,7 +37,7 @@
 IPAddress ClusterConnectionString::determineLocalSourceIP() const {
 	int size = coords.size() + hostnames.size();
 	int index = 0;
-	loop {
+	while (true) {
 		try {
 			using namespace boost::asio;
 

fdbclient/include/fdbclient/ClientWorkerInterface.h

@@ -22,6 +22,7 @@
 #define FDBCLIENT_CLIENTWORKERINTERFACE_H
 #pragma once
 
+#include "fdbrpc/FlowGrpc.h"
 #include "fdbclient/FDBTypes.h"
 #include "fdbrpc/FailureMonitor.h"
 #include "fdbclient/Status.h"
@@ -35,17 +36,24 @@ struct ClientWorkerInterface {
 	RequestStream<struct RebootRequest> reboot;
 	RequestStream<struct ProfilerRequest> profiler;
 	RequestStream<struct SetFailureInjection> setFailureInjection;
+	Optional<NetworkAddress> grpcAddress;
 
 	bool operator==(ClientWorkerInterface const& r) const { return id() == r.id(); }
 	bool operator!=(ClientWorkerInterface const& r) const { return id() != r.id(); }
 	UID id() const { return reboot.getEndpoint().token; }
 	NetworkAddress address() const { return reboot.getEndpoint().getPrimaryAddress(); }
 
-	void initEndpoints() { reboot.getEndpoint(TaskPriority::ReadSocket); }
+	void initEndpoints() {
+		reboot.getEndpoint(TaskPriority::ReadSocket);
+		auto grpcInstance = FlowGrpc::instance();
+		if (grpcInstance) {
+			grpcAddress = grpcInstance->server()->getAddress();
+		}
+	}
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, reboot, profiler, setFailureInjection);
+		serializer(ar, reboot, profiler, setFailureInjection, grpcAddress);
 	}
 };
 

fdbrpc/include/fdbrpc/Locality.h

@@ -410,16 +410,20 @@ struct ProcessData {
 	LocalityData locality;
 	ProcessClass processClass;
 	NetworkAddress address;
+	Optional<NetworkAddress> grpcAddress;
 
 	ProcessData() {}
-	ProcessData(LocalityData locality, ProcessClass processClass, NetworkAddress address)
-	  : locality(locality), processClass(processClass), address(address) {}
+	ProcessData(LocalityData locality,
+	            ProcessClass processClass,
+	            NetworkAddress address,
+	            Optional<NetworkAddress> grpcAddress)
+	  : locality(locality), processClass(processClass), address(address), grpcAddress(grpcAddress) {}
 
 	// To change this serialization, ProtocolVersion::WorkerListValue must be updated, and downgrades need to be
 	// considered
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, locality, processClass, address);
+		serializer(ar, locality, processClass, address, grpcAddress);
 	}
 
 	struct sort_by_address {

fdbserver/ClusterController.actor.cpp

@@ -916,8 +916,9 @@ ACTOR Future<Void> workerAvailabilityWatch(WorkerInterface worker,
 	state Future<Void> failed = (worker.address() == g_network->getLocalAddress())
 	                                ? Never()
 	                                : waitFailureClient(worker.waitFailure, SERVER_KNOBS->WORKER_FAILURE_TIME);
-	cluster->updateWorkerList.set(worker.locality.processId(),
-	                              ProcessData(worker.locality, startingClass, worker.stableAddress()));
+	cluster->updateWorkerList.set(
+	    worker.locality.processId(),
+	    ProcessData(worker.locality, startingClass, worker.stableAddress(), worker.grpcAddress()));
 	// This switching avoids a race where the worker can be added to id_worker map after the workerAvailabilityWatch
 	// fails for the worker.
 	wait(delay(0));

fdbserver/include/fdbserver/WorkerInterface.actor.h

@@ -85,6 +85,7 @@ struct WorkerInterface {
 	NetworkAddress stableAddress() const { return tLog.getEndpoint().getStableAddress(); }
 	Optional<NetworkAddress> secondaryAddress() const { return tLog.getEndpoint().addresses.secondaryAddress; }
 	NetworkAddressList addresses() const { return tLog.getEndpoint().addresses; }
+	Optional<NetworkAddress> grpcAddress() const { return clientInterface.grpcAddress; }
 
 	WorkerInterface() {}
 	WorkerInterface(const LocalityData& locality) : locality(locality) {}

fdbserver/worker.actor.cpp

@@ -82,7 +82,6 @@
 #include "flow/ApiVersion.h"
 #include "fdbcli_lib/CliService.h"
 
-
 #ifdef __linux__
 #include <fcntl.h>
 #include <stdio.h>