Discovery Service

Question: A discovery service is one of the most powerful applications a team can have at its disposal. A discovery service maintains information about every service in the application estate, and this information can be queried via a discovery service client API. This enables services to easily obtain location details of other services, and communicate directly with them. No more excessive hardcoded config with connection details is necessary. We will build a discovery service using an asynchronous heartbeat model. Every client will send an asynchronous heartbeat to the discovery service on an interval, which the discovery service will receive and store in a ‘status’ table and ‘latestStatus’ table. Status will maintain the history of heartbeats, whereas latestStatus will maintain the latest heartbeat. A client which disconnects is considered inactive, whereas a client that has sent a heartbeat in the past and has not disconnected is considered active. The discovery service can even heartbeat to itself and maintain its own information, which we will do as well in this implementation. The discovery service should have functions defined to return host and port information of active services, and the client API should call these functions remotely synchronously. The below solution demonstrates a basic and portable discovery service/client API implementation that can be applied to any kdb+ estate.

Example

                                
                                // Start up discovery service and have it load client API so it can heartbeat to itself 
$ q discoServer.q -p 7001 -discoPort 7001 -name discovery_1 -serviceType discovery -t 5000 -region amrs -sendHB 1 
q)\l discoClient.q 
 
// Start up client ‘services’ (empty q sessions with client API loaded in) 
$ q discoClient.q -p 8000 -discoHost localhost -discoPort 7001 -name rdb_amrs_1 -serviceType rdb -t 5000 -region amrs -sendHB 1 > /dev/null 2>&1 & 
$ q discoClient.q -p 8001 -discoHost localhost -discoPort 7001 -name rdb_amrs_2 -serviceType rdb -t 5000 -region amrs -sendHB 1 > /dev/null 2>&1 & 
$ q discoClient.q -p 8002 -discoHost localhost -discoPort 7001 -name hdb_amrs_1 -serviceType hdb -t 5000 -region amrs -sendHB 1 > /dev/null 2>&1 & 
$ q discoClient.q -p 8003 -discoHost localhost -discoPort 7001 -name hdb_amrs_2 -serviceType hdb -t 5000 -region amrs -sendHB 1 > /dev/null 2>&1 & 
$ q discoClient.q -p 8004 -discoHost localhost -discoPort 7001 -name hdb_amrs_3 -serviceType hdb -t 5000 -region amrs -sendHB 1 > /dev/null 2>&1 & 
$ q discoClient.q -p 9000 -discoHost localhost -discoPort 7001 -name rdb_emea_1 -serviceType rdb -t 5000 -region emea -sendHB 1 > /dev/null 2>&1 & 
 
// Back in discovery service 
q)latestStatus 
name       | time                          pid  serviceType region hostname        port qVersion qVersionDate ipcHandle active 
-----------| ----------------------------------------------------------------------------------------------------------------- 
discovery_1| 2020.10.08D19:26:02.423369000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
hdb_amrs_2 | 2020.10.08D19:26:05.274433000 1574 hdb         amrs   desktop-71jn2qg 8003 4        2020.06.01   6         1 
rdb_amrs_1 | 2020.10.08D19:26:05.274455000 1566 rdb         amrs   desktop-71jn2qg 8000 4        2020.06.01   7         1 
hdb_amrs_3 | 2020.10.08D19:26:05.282092000 1577 hdb         amrs   desktop-71jn2qg 8004 4        2020.06.01   8         1 
hdb_amrs_1 | 2020.10.08D19:26:05.282112000 1570 hdb         amrs   desktop-71jn2qg 8002 4        2020.06.01   9         1 
rdb_emea_1 | 2020.10.08D19:26:05.374412000 1578 rdb         emea   desktop-71jn2qg 9000 4        2020.06.01   10        1 
rdb_amrs_2 | 2020.10.08D19:26:05.380388000 1568 rdb         amrs   desktop-71jn2qg 8001 4        2020.06.01   11        1 
q)status 
name        time                          pid  serviceType region hostname        port qVersion qVersionDate ipcHandle active 
----------------------------------------------------------------------------------------------------------------------------- 
discovery_1 2020.10.08D19:25:07.423366000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
discovery_1 2020.10.08D19:25:12.422955000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
discovery_1 2020.10.08D19:25:17.423374000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
discovery_1 2020.10.08D19:25:22.423375000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
discovery_1 2020.10.08D19:25:27.423375000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
hdb_amrs_2  2020.10.08D19:25:30.274206000 1574 hdb         amrs   desktop-71jn2qg 8003 4        2020.06.01   6         1 
rdb_amrs_1  2020.10.08D19:25:30.274225000 1566 rdb         amrs   desktop-71jn2qg 8000 4        2020.06.01   7         1 
hdb_amrs_3  2020.10.08D19:25:30.281500000 1577 hdb         amrs   desktop-71jn2qg 8004 4        2020.06.01   8         1 
hdb_amrs_1  2020.10.08D19:25:30.281659000 1570 hdb         amrs   desktop-71jn2qg 8002 4        2020.06.01   9         1 
rdb_emea_1  2020.10.08D19:25:30.373906000 1578 rdb         emea   desktop-71jn2qg 9000 4        2020.06.01   10        1 
rdb_amrs_2  2020.10.08D19:25:30.379487000 1568 rdb         amrs   desktop-71jn2qg 8001 4        2020.06.01   11        1 
discovery_1 2020.10.08D19:25:32.423191000 1559 discovery   amrs   desktop-71jn2qg 7001 4        2020.06.01   0         1 
hdb_amrs_2  2020.10.08D19:25:35.274438000 1574 hdb         amrs   desktop-71jn2qg 8003 4        2020.06.01   6         1 
rdb_amrs_1  2020.10.08D19:25:35.274460000 1566 rdb         amrs   desktop-71jn2qg 8000 4        2020.06.01   7         1 
hdb_amrs_1  2020.10.08D19:25:35.282184000 1570 hdb         amrs   desktop-71jn2qg 8002 4        2020.06.01   9         1 
.. 
 
// Open up new q session with client API loaded in and test API functions 
$ q discoClient.q -discoHost localhost -discoPort 7001 
q).discoClient.getServiceAll (1#`region)!1#`amrs 
`:desktop-71jn2qg:7001`:desktop-71jn2qg:8003`:desktop-71jn2qg:8000`:desktop-71jn2qg:8004`:desktop-71jn2qg:8002`:desktop-71jn2qg:8001 
q).discoClient.getServiceAll `region`serviceType!`amrs`rdb 
`:desktop-71jn2qg:8000`:desktop-71jn2qg:8001 
q).discoClient.getServiceAll `region`serviceType!`amrs`hdb 
`:desktop-71jn2qg:8003`:desktop-71jn2qg:8004`:desktop-71jn2qg:8002 
q).discoClient.getServiceAll `region`serviceType!`emea`hdb 
`symbol$() 
q).discoClient.getService `region`serviceType!`amrs`hdb 
`:desktop-71jn2qg:8003 
q).discoClient.getService `region`serviceType!`amrs`hdb 
`:desktop-71jn2qg:8002 
q).discoClient.getService `region`serviceType!`emea`hdb 
` 
q).discoClient.getServiceAll (1#`name)!enlist"rdb_amrs*" 
`:desktop-71jn2qg:8000`:desktop-71jn2qg:8001 
q).discoClient.getServiceAll `serviceType`qVersion!(`rdb;4) 
`:desktop-71jn2qg:8000`:desktop-71jn2qg:9000`:desktop-71jn2qg:8001

Solution

                                
                                ######## discoClient.q ########

\d .discoClient

discoHdl:0Ni;

args:.Q.def[`name`serviceType`region`discoHost`discoPort`sendHB!(`;`;`;`;0Ni;0b)] .Q.opt .z.x;

connect:{ 
    conn:`;
    if[null args[`discoPort]; '"No discovery port specified"];
    $[(args[`discoHost]=.z.h) or (null args[`discoHost]) or args[`discoHost]=`localhost;
         conn:.Q.dd[`:unix:/;args[`discoPort]];
         conn:hsym `$":" sv string (args[`discoHost];args[`discoPort]);
    ];
    h:@[hopen;(conn;1000);{discoHdl::0Ni;'x}];
    discoHdl::h;
 };

disconnect:{ 
    @[hclose;discoHdl;()];
    discoHdl::0Ni;
 };

sendUpd:{
    d:`name`pid`serviceType`region`hostname`port`qVersion`qVersionDate!(args`name;.z.i;args`serviceType;args`region;.z.h;system"p";.z.K;.z.k);
    neg[discoHdl](`upd;d);
 };

// get host and port for service(s) requested, pass in dictionary, ex. `serviceType`region!`rdb`amrs
getServiceAll:{ discoHdl(`getServiceAll;x)};
getService:{ discoHdl(`getService;x)};

.z.pc:{
    if[x=discoHdl;
        disconnect[];
    ];
 };

// if disconnected try to reconnect on next timer cycle
.z.ts:{
    if[null discoHdl;connect[]];
    if[args[`sendHB] and not null discoHdl; sendUpd[]];
 };

connect[];

######## discoServer.q ########

status:([]
    name:`$();
    time:`timestamp$();
    pid:`int$();
    serviceType:`$();
    region:`$();
    hostname:`$();
    port:`int$();
    qVersion:`float$();
    qVersionDate:`date$();
    ipcHandle:`int$();
    active:`boolean$()
 );

latestStatus:1!status;

upd:{ 
    time:.z.p;
    // new ipc handle for calling service, close old one
    if[not[null oldH] and not .z.w~oldH:latestStatus[x`name]`ipcHandle;
        `status upsert @[;`active;:;0b] @[;`time;:;time] exec from status where ipcHandle=oldH;
        @[hclose;oldH;()];
    ];
    row:(x`name;time;x`pid;x`serviceType;x`region;x`hostname;x`port;x`qVersion;x`qVersionDate;.z.w;1b);
    `status upsert row;
    `latestStatus upsert row;
 };

// check disconnected handle against latestStatus and update tables if necessary
.z.pc:{
    if[not[null inactive] and count inactive:first exec name from latestStatus where ipcHandle=x;
        update ipcHandle:0Ni, active:0b from `latestStatus where name=inactive;
        `status upsert @[;`time;:;.z.p] exec from latestStatus where name=inactive;
    ];
 };

// only accept function calls from sync messages
ALLOWED_FNS:`getService`getServiceAll;
.z.pg:{
    if[10h=type x;x:parse x];
    if[not -11h=type fn:x 0;'"Not allowed"];
    if[not fn in ALLOWED_FNS;'"Not allowed"];
    value x
 };

// only accept heartbeats from async messages
.z.ps:{
    if[10h=type x;x:parse x];
    if[not `upd~x 0;'"Not allowed"];
    value x;
 };

createHostPort:{[h;p]
    if[not[count h] or not count p;:`];
    if[all[null h] or all null p;:`];
    hp:hsym `$(":" sv) each string ((),h),'(),p;
    $[1=count hp;first hp;hp]
 };

// discovery functions
getServiceAll:{[opts]
    allowedFilters:`name`serviceType`region`hostname`port`qVersion`qVersionDate;
    opts:(key[opts] inter allowedFilters)#opts;
    if[not count opts;:`$()];
    // select from latestStatus where active, not ipcHandle=.z.w, ...
    filters:(`active;(not;(=;`ipcHandle;`.z.w))),key[opts] {fn:(=;like) x=`name;(fn;x;$[-11h=type y;1#;(::)] y)}' value[opts];
    svcs:?[`latestStatus;filters;0b;()];
    except[;`] (),exec createHostPort[hostname;port] from svcs
 };
getService:{[opts] first -1?getServiceAll[opts] };

Explanation: For the discovery service, the status and latestStatus tables need to be created. The latestStatus table is a keyed variant of status, and it groups services by their name. This means that every service should have a unique name. A 'upd' function needs to be defined (can be called something else) for the clients to call remotely asynchronously. The upd function will insert the client information into the status table and latestStatus table. The IPC handle close callback .z.pc is defined and updates the status and latestStatus table when a client disconnects. It updates the active flag to false in latestStatus for the client and creates a new entry in status with the active flag set to false. The IPC get handlers .z.pg/.z.ps are defined and only accept allowed messages from outside q processes. For synchronous messages, outsiders can only call the functions 'getService' and 'getServiceAll'. For asynchronous messages, outsiders can only send heartbeats. The functions 'getService' and 'getServiceAll' are defined and take in a dictionary argument which specifies the filters to apply, and queries the client information from latestStatus while applying the filters. The functions return the host and port information of the clients, with getServiceAll returning all information and getService returning only one client at random.<br> For the client API, the connect/disconnect/heartbeat functions need to be defined. The vital pieces of information included in the heartbeat messages are passed into the q process through commandline arguments. The arguments 'discoHost' and 'discoPort' refer to the host/port information of the discovery service. If 'discoHost' is null, then it is assumed the discovery service lives on the same host, and a Unix Domain Socket connection is used instead of TCP. The functions 'getServiceAll' and 'getService' call the corresponding functions remotely synchronously on the discovery service. The IPC handle close callback is defined and calls disconnect when the discovery service connection is dropped abruptly. The timer function .z.ts is defined and attemps reconnection if the process is disconnected from the discovery service, and sends heartbeats to the discovery service if it is connected.

Code:

https://bitbucket.org/alvikabir/dailyq/src/master/Advanced/discovery-service

Tags:

api architecture asynchronous ipc

Searchable Tags

algorithms api architecture asynchronous c csv data structures dictionaries disk feedhandler finance functions ingestion ipc iterators machine learning math multithreading optimizations realtime shared library sql statistics streaming strings tables temporal utility websockets

daily q

Discovery Service

Example

Solution

Tags:

Searchable Tags