Skip to content

Commit

Permalink
feat: added nodes connections command to list active NodeConnections
Browse files Browse the repository at this point in the history
  • Loading branch information
tegefaulkes committed Nov 10, 2022
1 parent ab8d759 commit ff2d436
Show file tree
Hide file tree
Showing 13 changed files with 622 additions and 16 deletions.
87 changes: 87 additions & 0 deletions src/bin/nodes/CommandConnections.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
import type PolykeyClient from '../../PolykeyClient';
import type nodesPB from '../../proto/js/polykey/v1/nodes/nodes_pb';
import CommandPolykey from '../CommandPolykey';
import * as binUtils from '../utils/utils';
import * as binProcessors from '../utils/processors';

class CommandAdd extends CommandPolykey {
constructor(...args: ConstructorParameters<typeof CommandPolykey>) {
super(...args);
this.name('connections');
this.description('list all active node connections');
this.action(async (options) => {
const { default: PolykeyClient } = await import('../../PolykeyClient');
const utilsPB = await import('../../proto/js/polykey/v1/utils/utils_pb');
const clientOptions = await binProcessors.processClientOptions(
options.nodePath,
options.nodeId,
options.clientHost,
options.clientPort,
this.fs,
this.logger.getChild(binProcessors.processClientOptions.name),
);
const meta = await binProcessors.processAuthentication(
options.passwordFile,
this.fs,
);
let pkClient: PolykeyClient;
this.exitHandlers.handlers.push(async () => {
if (pkClient != null) await pkClient.stop();
});
try {
pkClient = await PolykeyClient.createPolykeyClient({
nodePath: options.nodePath,
nodeId: clientOptions.nodeId,
host: clientOptions.clientHost,
port: clientOptions.clientPort,
logger: this.logger.getChild(PolykeyClient.name),
});
// DO things here...
// Like create the message.
const emptyMessage = new utilsPB.EmptyMessage();

const connections = await binUtils.retryAuthentication(async (auth) => {
const connections = pkClient.grpcClient.nodesListConnections(
emptyMessage,
auth,
);
const connectionEntries: Array<nodesPB.NodeConnection.AsObject> = [];
for await (const connection of connections) {
connectionEntries.push(connection.toObject());
}
return connectionEntries;
}, meta);
if (options.format === 'human') {
const output: Array<string> = [];
for (const connection of connections) {
const hostnameString =
connection.hostname === '' ? '' : `(${connection.hostname})`;
const hostString = `${connection.nodeId}@${connection.host}${hostnameString}:${connection.port}`;
const usageCount = connection.usageCount;
const timeout =
connection.timeout === -1 ? 'NA' : `${connection.timeout}`;
const outputLine = `${hostString}\t${usageCount}\t${timeout}`;
output.push(outputLine);
}
process.stdout.write(
binUtils.outputFormatter({
type: 'list',
data: output,
}),
);
} else {
process.stdout.write(
binUtils.outputFormatter({
type: 'json',
data: connections,
}),
);
}
} finally {
if (pkClient! != null) await pkClient.stop();
}
});
}
}

export default CommandAdd;
2 changes: 2 additions & 0 deletions src/bin/nodes/CommandNodes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import CommandClaim from './CommandClaim';
import CommandFind from './CommandFind';
import CommandPing from './CommandPing';
import CommandGetAll from './CommandGetAll';
import CommandConnections from './CommandConnections';
import CommandPolykey from '../CommandPolykey';

class CommandNodes extends CommandPolykey {
Expand All @@ -15,6 +16,7 @@ class CommandNodes extends CommandPolykey {
this.addCommand(new CommandFind(...args));
this.addCommand(new CommandPing(...args));
this.addCommand(new CommandGetAll(...args));
this.addCommand(new CommandConnections(...args));
}
}

Expand Down
19 changes: 19 additions & 0 deletions src/client/GRPCClientClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -915,6 +915,25 @@ class GRPCClientClient extends GRPCClient<ClientServiceClient> {
)(...args);
}

@ready(new clientErrors.ErrorClientClientDestroyed())
public nodesListConnections(
...args
): AsyncGeneratorReadableStreamClient<
nodesPB.NodeConnection,
ClientReadableStream<nodesPB.NodeConnection>
> {
return grpcUtils.promisifyReadableStreamCall<nodesPB.NodeConnection>(
this.client,
{
nodeId: this.nodeId,
host: this.host,
port: this.port,
command: this.nodesListConnections.name,
},
this.client.nodesListConnections,
)(...args);
}

@ready(new clientErrors.ErrorClientClientDestroyed())
public identitiesAuthenticate(...args) {
return grpcUtils.promisifyReadableStreamCall<identitiesPB.AuthenticationProcess>(
Expand Down
2 changes: 2 additions & 0 deletions src/client/service/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ import nodesClaim from './nodesClaim';
import nodesFind from './nodesFind';
import nodesPing from './nodesPing';
import nodesGetAll from './nodesGetAll';
import nodesListConnections from './nodesListConnections';
import notificationsClear from './notificationsClear';
import notificationsRead from './notificationsRead';
import notificationsSend from './notificationsSend';
Expand Down Expand Up @@ -167,6 +168,7 @@ function createService({
nodesFind: nodesFind(container),
nodesPing: nodesPing(container),
nodesGetAll: nodesGetAll(container),
nodesListConnections: nodesListConnections(container),
notificationsClear: notificationsClear(container),
notificationsRead: notificationsRead(container),
notificationsSend: notificationsSend(container),
Expand Down
52 changes: 52 additions & 0 deletions src/client/service/nodesListConnections.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
import type * as grpc from '@grpc/grpc-js';
import type { Authenticate } from '../types';
import type * as utilsPB from '../../proto/js/polykey/v1/utils/utils_pb';
import type Logger from '@matrixai/logger';
import type NodeConnectionManager from '../../nodes/NodeConnectionManager';
import * as grpcUtils from '../../grpc/utils';
import * as nodesPB from '../../proto/js/polykey/v1/nodes/nodes_pb';
import * as clientUtils from '../utils';
import * as nodesUtils from '../../nodes/utils';

function nodesListConnections({
authenticate,
nodeConnectionManager,
logger,
}: {
authenticate: Authenticate;
nodeConnectionManager: NodeConnectionManager;
logger: Logger;
}) {
return async (
call: grpc.ServerWritableStream<
utilsPB.EmptyMessage,
nodesPB.NodeConnection
>,
): Promise<void> => {
const genWritable = grpcUtils.generatorWritable(call, false);
try {
const metadata = await authenticate(call.metadata);
call.sendMetadata(metadata);
const connections = nodeConnectionManager.listConnections();
for (const connection of connections) {
const connectionMessage = new nodesPB.NodeConnection();
connectionMessage.setNodeId(nodesUtils.encodeNodeId(connection.nodeId));
connectionMessage.setHost(connection.address.host);
connectionMessage.setHostname(connection.address.hostname ?? '');
connectionMessage.setPort(connection.address.port);
connectionMessage.setUsageCount(connection.usageCount);
connectionMessage.setTimeout(connection.timeout ?? -1);
await genWritable.next(connectionMessage);
}
await genWritable.next(null);
return;
} catch (e) {
await genWritable.throw(e);
!clientUtils.isClientClientError(e) &&
logger.error(`${nodesListConnections.name}:${e}`);
return;
}
};
}

export default nodesListConnections;
34 changes: 34 additions & 0 deletions src/nodes/NodeConnectionManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1028,10 +1028,44 @@ class NodeConnectionManager {
return establishedMap;
}

@ready(new nodesErrors.ErrorNodeConnectionManagerNotRunning())
public hasConnection(nodeId: NodeId): boolean {
return this.connections.has(nodeId.toString() as NodeIdString);
}

@ready(new nodesErrors.ErrorNodeConnectionManagerNotRunning())
public listConnections(): Array<{
nodeId: NodeId;
address: { host: Host; port: Port; hostname: Hostname | undefined };
usageCount: number;
timeout: number | undefined;
}> {
const results: Array<{
nodeId: NodeId;
address: { host: Host; port: Port; hostname: Hostname | undefined };
usageCount: number;
timeout: number | undefined;
}> = [];
for (const [
nodeIdString,
connectionAndTimer,
] of this.connections.entries()) {
const connection = connectionAndTimer.connection;
const nodeId = IdInternal.fromString<NodeId>(nodeIdString);
results.push({
nodeId,
address: {
host: connection.host,
port: connection.port,
hostname: connection.hostname,
},
usageCount: connectionAndTimer.usageCount,
timeout: connectionAndTimer.timer?.getTimeout(),
});
}
return results;
}

protected hasBackoff(nodeId: NodeId): boolean {
const backoff = this.nodesBackoffMap.get(nodeId.toString());
if (backoff == null) return false;
Expand Down
15 changes: 15 additions & 0 deletions src/proto/js/polykey/v1/client_service_grpc_pb.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ interface IClientServiceService extends grpc.ServiceDefinition<grpc.UntypedServi
nodesClaim: IClientServiceService_INodesClaim;
nodesFind: IClientServiceService_INodesFind;
nodesGetAll: IClientServiceService_INodesGetAll;
nodesListConnections: IClientServiceService_INodesListConnections;
keysKeyPairRoot: IClientServiceService_IKeysKeyPairRoot;
keysKeyPairReset: IClientServiceService_IKeysKeyPairReset;
keysKeyPairRenew: IClientServiceService_IKeysKeyPairRenew;
Expand Down Expand Up @@ -167,6 +168,15 @@ interface IClientServiceService_INodesGetAll extends grpc.MethodDefinition<polyk
responseSerialize: grpc.serialize<polykey_v1_nodes_nodes_pb.NodeBuckets>;
responseDeserialize: grpc.deserialize<polykey_v1_nodes_nodes_pb.NodeBuckets>;
}
interface IClientServiceService_INodesListConnections extends grpc.MethodDefinition<polykey_v1_utils_utils_pb.EmptyMessage, polykey_v1_nodes_nodes_pb.NodeConnection> {
path: "/polykey.v1.ClientService/NodesListConnections";
requestStream: false;
responseStream: true;
requestSerialize: grpc.serialize<polykey_v1_utils_utils_pb.EmptyMessage>;
requestDeserialize: grpc.deserialize<polykey_v1_utils_utils_pb.EmptyMessage>;
responseSerialize: grpc.serialize<polykey_v1_nodes_nodes_pb.NodeConnection>;
responseDeserialize: grpc.deserialize<polykey_v1_nodes_nodes_pb.NodeConnection>;
}
interface IClientServiceService_IKeysKeyPairRoot extends grpc.MethodDefinition<polykey_v1_utils_utils_pb.EmptyMessage, polykey_v1_keys_keys_pb.KeyPair> {
path: "/polykey.v1.ClientService/KeysKeyPairRoot";
requestStream: false;
Expand Down Expand Up @@ -684,6 +694,7 @@ export interface IClientServiceServer extends grpc.UntypedServiceImplementation
nodesClaim: grpc.handleUnaryCall<polykey_v1_nodes_nodes_pb.Claim, polykey_v1_utils_utils_pb.StatusMessage>;
nodesFind: grpc.handleUnaryCall<polykey_v1_nodes_nodes_pb.Node, polykey_v1_nodes_nodes_pb.NodeAddress>;
nodesGetAll: grpc.handleUnaryCall<polykey_v1_utils_utils_pb.EmptyMessage, polykey_v1_nodes_nodes_pb.NodeBuckets>;
nodesListConnections: grpc.handleServerStreamingCall<polykey_v1_utils_utils_pb.EmptyMessage, polykey_v1_nodes_nodes_pb.NodeConnection>;
keysKeyPairRoot: grpc.handleUnaryCall<polykey_v1_utils_utils_pb.EmptyMessage, polykey_v1_keys_keys_pb.KeyPair>;
keysKeyPairReset: grpc.handleUnaryCall<polykey_v1_keys_keys_pb.Key, polykey_v1_utils_utils_pb.EmptyMessage>;
keysKeyPairRenew: grpc.handleUnaryCall<polykey_v1_keys_keys_pb.Key, polykey_v1_utils_utils_pb.EmptyMessage>;
Expand Down Expand Up @@ -770,6 +781,8 @@ export interface IClientServiceClient {
nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, options: Partial<grpc.CallOptions>, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
nodesListConnections(request: polykey_v1_utils_utils_pb.EmptyMessage, options?: Partial<grpc.CallOptions>): grpc.ClientReadableStream<polykey_v1_nodes_nodes_pb.NodeConnection>;
nodesListConnections(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata?: grpc.Metadata, options?: Partial<grpc.CallOptions>): grpc.ClientReadableStream<polykey_v1_nodes_nodes_pb.NodeConnection>;
keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, options: Partial<grpc.CallOptions>, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
Expand Down Expand Up @@ -958,6 +971,8 @@ export class ClientServiceClient extends grpc.Client implements IClientServiceCl
public nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
public nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
public nodesGetAll(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, options: Partial<grpc.CallOptions>, callback: (error: grpc.ServiceError | null, response: polykey_v1_nodes_nodes_pb.NodeBuckets) => void): grpc.ClientUnaryCall;
public nodesListConnections(request: polykey_v1_utils_utils_pb.EmptyMessage, options?: Partial<grpc.CallOptions>): grpc.ClientReadableStream<polykey_v1_nodes_nodes_pb.NodeConnection>;
public nodesListConnections(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata?: grpc.Metadata, options?: Partial<grpc.CallOptions>): grpc.ClientReadableStream<polykey_v1_nodes_nodes_pb.NodeConnection>;
public keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
public keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
public keysKeyPairRoot(request: polykey_v1_utils_utils_pb.EmptyMessage, metadata: grpc.Metadata, options: Partial<grpc.CallOptions>, callback: (error: grpc.ServiceError | null, response: polykey_v1_keys_keys_pb.KeyPair) => void): grpc.ClientUnaryCall;
Expand Down
22 changes: 22 additions & 0 deletions src/proto/js/polykey/v1/client_service_grpc_pb.js
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,17 @@ function deserialize_polykey_v1_nodes_NodeBuckets(buffer_arg) {
return polykey_v1_nodes_nodes_pb.NodeBuckets.deserializeBinary(new Uint8Array(buffer_arg));
}

function serialize_polykey_v1_nodes_NodeConnection(arg) {
if (!(arg instanceof polykey_v1_nodes_nodes_pb.NodeConnection)) {
throw new Error('Expected argument of type polykey.v1.nodes.NodeConnection');
}
return Buffer.from(arg.serializeBinary());
}

function deserialize_polykey_v1_nodes_NodeConnection(buffer_arg) {
return polykey_v1_nodes_nodes_pb.NodeConnection.deserializeBinary(new Uint8Array(buffer_arg));
}

function serialize_polykey_v1_notifications_List(arg) {
if (!(arg instanceof polykey_v1_notifications_notifications_pb.List)) {
throw new Error('Expected argument of type polykey.v1.notifications.List');
Expand Down Expand Up @@ -590,6 +601,17 @@ nodesAdd: {
responseSerialize: serialize_polykey_v1_nodes_NodeBuckets,
responseDeserialize: deserialize_polykey_v1_nodes_NodeBuckets,
},
nodesListConnections: {
path: '/polykey.v1.ClientService/NodesListConnections',
requestStream: false,
responseStream: true,
requestType: polykey_v1_utils_utils_pb.EmptyMessage,
responseType: polykey_v1_nodes_nodes_pb.NodeConnection,
requestSerialize: serialize_polykey_v1_utils_EmptyMessage,
requestDeserialize: deserialize_polykey_v1_utils_EmptyMessage,
responseSerialize: serialize_polykey_v1_nodes_NodeConnection,
responseDeserialize: deserialize_polykey_v1_nodes_NodeConnection,
},
// Keys
keysKeyPairRoot: {
path: '/polykey.v1.ClientService/KeysKeyPairRoot',
Expand Down
Loading

0 comments on commit ff2d436

Please sign in to comment.