Skip to content

Service Gateway interface

There are two types of processes that can interact with the SG: client processes making API calls, and data access processes (DAPs) servicing requests (e.g. RDB/IDB/HDB).

Client

To make an API call to the SG from a q process, do the following: 1. hopen a connection to a GW process. 2. Issue an API call by passing a 4-element list with the following elements:

  • apiName - symbol - API name.
  • args - dictionary - Named arguments to the API. These can include:
    • table - symbol - Table name.
    • startTS - timestamp - Start time (inclusive) -- used for routing.
    • endTS - timestamp - End time (exclusive) -- used for routing.
    • inputTZ - symbol - Timezone of startTS and endTS.
    • outputTZ - symbol - Timezone of API response.
    • DAP labels -- used for routing.
    • API-specific arguments.
  • callback -symbol - Dyadic callback function to be invoked on request completion in the async case (see below).
  • opts - dictionary - Optional out-of-band information to accompany the request (see "header" page). This MUST be an untyped dictionary. For no opts, use (0#`)!().

API requests are done synchronously or asynchronously. Upon completion of the request, the GW uses the hopened handle to invoke the callback function with the following arguments:

  • hdr - dictionary - Response header (see "header" page).
  • payload - any - Response payload.

Example

h:hopen`:gwHost:gwPort;
myCallbackFn:{[hdr;payload] ...};

/ Async -- `myCallBackFn` will be invoked.
neg[h](`myAPI;`startTS`endTS`region`assetClass`arg1`arg2!(-0Wp;0Wp;`amer;`equity;1;"abc");`myCallbackFn;`logCorr`appCorr!("request_1";10));

/ Sync -- response is a two-element list: (hdr;payload).
res:h(`myAPI;`startTS`endTS`region`assetClass`arg1`arg2!(-0Wp;0Wp;`amer;`equity;1;"abc");`myCallbackFn;`logCorr`appCorr!("request_1";10));
res 0 / Header
res 1 / Payload

Moreover, note that, whether the API itself uses the arguments or not, all API calls may includes startTS, endTS and any DAP labels. If not included, these default to "all", i.e.

arg default
startTS -0Wp
endTS 0Wp
labels All available labels

For example, suppose DAPs are labeled by exchange and assetClass, and the list of available labels is:

exchange assetClass
nyse equity
nyse futures
tsx equity

Some examples of defaulting:

  • No routing args is equivalent to querying for all available exchanges/asset classes/time, i.e.

    exchange assetClass startTS endTS
    nyse equity -0Wp 0Wp
    nyse futures -0Wp 0Wp
    tsx equity -0Wp 0Wp

  • If specifying exchange=nyse and startTS=2021.01.31D, then assetClass defaults to `equity`futures and endTS defaults to 0Wp.

  • If specifying exchange=futures and endTS=2021.10.06D, then assetClass defaults to `nyse (as there is no tsx-futures and startTS defaults to -0Wp.

Discovery

If your client process uses KX Insights Discovery, Gateway processes can be found via discovery. The following sample code will select all active GWs:

sn:`$"KXI-SG-GW" / GW service name
select from .com_kx_sd.getServices[]where serviceName=sn,status=`UP

REST

Optionally, API requests for the pre-configured APIs (see "API" page) can be made via the REST interface if it is configured to run (see "Configuration") and the DAPs implement the APIs in question (i.e. you are using the KX Insights Data Access microservcie, or your own custom DAPs with your implementations of the APIs -- see "API" page for details).

Data Access Process

In order to integrate with the GW microservice, a DAP must do the following (all functions are documented below and a sample example is provided):

  • Register with the Resource Coordinator. This can be done by opening a connection to the Resource Coordinator (hopen) and invoking .sgrc.registerDAP with its data purview, metadata, schemas and partitions (see Purview, Metadata, Schema and Partitions, respectively). The purview and partitions can be updated by calling .sgrc.updDapStatus.
  • On a failed DAP registration or update to status, the Resource Coordinator will call the function .da.registrationErr on the DAP with a response header that describes the error.
  • Execute APIs invoked by the client. The GW asynchronously invokes the following function on the DAP, which must be defined:
  • .da.execute - Executes an API.
    • apiName - symbol - API name.
    • hdr - dictionary - Request header (see "header" page).
    • args - dictionary - Named API arguments.

In response to this, the DAP must:

  • Send a response header and a response payload to the correct Aggregator by invoking .sgagg.onPartial on the host/port specified by hdr.agg.
  • Send a message to the Resource Coordinator indicating that it is available to do more work by invoking .sgrc.onPartial on the Resource Coordinator with an appropriate response header.

Note

The DAP is expected to respond even in the event of an error. Proper error handling should be implemented in order to guarantee this.

If using the KX Insights Data Access microservice, all of the above is automatically implemented; the DA only needs to be configured with its purview and the name of the SG that it should connect to (see "Data Access" configuration).

Purview and Routing

The purview describes the portion of the data each individual DAP offers and it is the means by which the SG makes its routing decisions. The purview is a dictionay with the following keys:

  • ver - long - Purview version number (used for keeping track of purview changes, not routing)
  • startTS - timestamp - Start time covered by DAP (inclusive).
  • endTS - timestamp - End time covered by DAP (exclusive).
  • Any other label values the SG should use in making routing decisions.

DAPs register and update their purviews with the RC using the APIs described below, and the RC makes routing decisions based on request args vs. DAP purviews. If the request args contain the table key, then table properties are taken into account. The routing logic is summarized in the following table.

isPartitioned isSharded routing strategy
true true Route to each distinct purview across time.
false true Route to one DAP per distinct purview.
false false Route to any one DAP.
true false Not supported.

If the table is unspecified, then the routing defaults to the isPartitioned=true and isSharded=true case.

For example, suppose DAPs are labelled by city and sensorType, and the registered DAPs are

dap city sensorType startTS endTS
dap1 toronto gas -0Wp 0Wp
dap2 toronto electric -0Wp 0Wp
dap3 montreal gas -0Wp 2021.06.01D
dap3 montreal gas 2021.05.01D 0Wp
dap4 montreal electric -0Wp 0Wp
dap5 vancouver gas -0Wp 0Wp
dap6 vancouver electric -0Wp 0Wp

  • If we receive a request for a partitioned table, or the table is not specified, with args `startTS`endTS`city`sensorType!(2021.05.10D;2021.06.15D;`toronto`montreal;`gas), then the RC sends the request to all relevant labels and across time as follows:

    dap args
    dap1 `startTS`endTS`city`sensorType!(2021.05.10D;2021.06.15D;`toronto;`gas)
    dap3 `startTS`endTS`city`sensorType!(2021.05.10D;2021.06.01D;`montreal;`gas)
    dap4 `startTS`endTS`city`sensorType!(2021.06.01D;2021.06.15D;`montreal;`gas)

  • If we receive a request for a sharded, non-partitioned table (i.e. a non-timeseries table that is distributed across assemblies), with args `city`sensorType!(`toronto`montreal;`gas), then the RC splits the request across relevant DAPs, but ignores time:

    dap args
    dap1 `city`sensorType!`toronto`gas
    dap3 or dap4 (but not both) `city`sensorType!`montreal`gas

  • If we receive a request for a non-sharded table (i.e. a table that is identical across all assemblies), with args `city`sensorType!(`toronto`montreal;`gas), then the RC sends to either dap1, dap3, or dap4, but only one of them.

Things to note:

  • Any missing purview args not specified default to "all". In the case of time, the defaults are startTS=-0Wp and endTS=0Wp.
  • Any additional arguments (not related to purview) are sent unchanged to all DAPs participating in the request.
  • The SG will not send the same portion of a request to two different DAPs if they overlap.
  • Once the SG enlists a DAP to service request, it is marked unavailable until it reports that it has completed (.sgrc.onPartial). If no DAPs are available to service all or part of a request, the SG sends out what it can, and enqueues the portions that can't be completed until a DAP registers or updates itself as being able to satisfy (wholly or partially) the outstanding portion of the request.
  • If different DAPs register with different sets of labels, the SG fills in missing labels with nulls.
  • A DAP must register with at least one label.

On the Aggregator

With the exception of the pre-configured APIs (see "API" page), the only supported aggregation function is raze. This will be expanded upon in future versions. Thus, all API results returned to the aggregator should be "raze-able" (i.e. raze(res 0; res 1;...)).

Metadata

DAPs may register with metadata information on the APIs it offers, which is subsequently surfaced through the .kxi.getMeta API (see "API" page). Metadata is a table with the following columns:

column type description example
fn symbol Function/API name. `myAPI
custom boolean Whether this is a custom (true) or built-in (false) API. 0b
params dictionary[] Parameters to the API (see "SAPI - metadata"). See "SAPI - metadata"
return dictionary Return of the API (see "SAPI - metadata"). See "SAPI - metadata"
misc dictionary Miscellaneous metadata fields. `x`y!("a";1)

Note that the custom field is traditionally used in the KX Insights DA microservice to distinguish between built-in KXI APIs and user-defined custom APIs. In the case that you are implementing your own DAP, this field can be used at your discretion.

Schema

DAPs may register with schema information, which is then subsequently surfaced through the .kxi.getMeta API (see "API" page). The schema is a table with the following columns:

column type description example
table symbol Table name. `trade
typ symbol Table type. `partitioned
pkCols symbol[] Primary key columns. `sym`time
updTsCol symbol Update time column. `updTime
prtnCol symbol Column the data is partitioned on. `time
sortColsMem symbol[] Columns the data is sorted on in memory (e.g. RDB). ,`updTime
sortColsIDisk symbol[] Columns the data is sorted on on-intraday-disk (e.g. IDB). `sym`updTime
sortColsDisk symbol[] Columns the data is sorted on on-disk (e.g. HDB). `sym`time
isPartitioned boolean Specifies if the table is partitioned. 1b
isSharded boolean Specifies if the table is sharded across multiple assemblies. 0b
columns table Table columns (see below). See below.

Each columns cell is a table with the following columns:

column type description example
column symbol Column name. `price
typ short KDB+ type. 9h
attrMem symbol In-memory attribute (e.g. RDB). `
attrIDisk symbol On-intraday-disk attribute (e.g. IDB). `s
attrDisk symbol On-disk attribute (e.g. HDB). `p
isSerialized boolean Whether column is serialized. 0b
fk symbol Foreign table information. instrument.sym

Note that the schema schema is strictly enforced when registering with the RC (see .sgrc.registerDAP), but the columns schema in individual cells is not.

Partitions

DAPs must register with partition information. Currently, partitioning is assumed to be by date. A DAP's partitions are given as a table with one of the following schemas:

column type description
date date Date of the partition.
startTS timestamp Inclusive start time of data within the partition.
endTS timestamp Exclusive end time of data within the partition.

OR

column type description
min_date date Inclusive minimum partition date.
max_date date Inclusive maximum partition date.

Note that in the latter schema, each date partition is assumed to cover the entire date in question.

Discovery

The DAP can use discovery to find the RC it should report to. If the DAP uses discovery, the following sample code can be used to discover the RC:

name:`myRC; / RC's KXI_NAME environment variable
sn:`$"KXI-SG-RC"; / RC service name
select from .com_kx_sd.getServices[]where serviceName=sg,status=`UP,name=metadata@\:`name
Note that the uniqueness of the RC process guarantees that there should only be one result from this select.

APIs

Below are the processes a DAP can connect to and the APIs that it can call.

  • Resource Coordinator

    • .sgrc.registerDAP - Registers DAP as an API target to service requests.
      • hp - symbol - Host port in the form `:host:1234.
      • avail - boolean - Flag indicating if the DAP is available to service requests or not.
      • purview - dictionary - See Purview.
      • asm - symbol - Assembly name -- used to identify a family of like DAPs.
      • instance - symbol - Instance type (e.g. RDB, IDB, HDB).
      • metadata - table|() - Metadata table (see Metadata) or an empty list to provide no metadata.
      • schema - table|() - Schema table (see Schema).
      • prtns - table - Partition information (see Partitions).

    Providing no metadata or schema information has no ill effects on query execution, however, the DAP/assembly will come up empty on a .kxi.getMeta call (see "API" page). Note that it is assumed that all DAPs within the same assembly (asm) share the same table schemas.

    • .sgrc.updDapStatus - Updates current target status.
      • avail - boolean - Flag indicating if the DAP is available to service requests or not.
      • purview - dictionary - See Purview.
      • prtns - table - See Partitions.

Note that certain subsets of the purview keys are allowed, for efficiency. The allowed combinations are: - All the keys specified above. - Only ver, startTS, and endTS, if no other purview dimensions have changed. - None, if the purview hasn't changed, and we only want to update availability. - .sgrc.onPartial - Informs the RC that we are available for more work following an API request. - hdr - dictionary - Response header (See "header" page). Must contain, at minimum, the following keys: - rc - byte - Response code. - ac - byte - Application code. - All other keys present in the request header.

Additionally, the DAP may use this notification to inform the RC that it was unable to send its response to the Aggregator. This is done by including the following key in the response header: - sendErr - any - Value is irrelevant.

If the sendErr key is present in the response header and hdr.rc is non-zero, the RC attempts to notify the Aggregator/GW of the failure in order to issue a response to the client.

  • Aggregator
    • .sgagg.onPartial - Forwards this DAPs partial results to the Aggregator for aggregation. This function should be executed on the process pointed to by agg in the request header (this request header value is a symbol of the form `:host:port).
      • hdr - dictionary - Response header (see "header" page). Must contain, at minimum, the following keys:
        • rc - short - Response code.
        • ac - short - Application code.
        • All other keys present in the request header.
      • payload - any - Response payload.

Example

The following is a very simple example of a DAP implementation.

//
// Purview - a DAP covers one possible combination of the SG's labels, and some temporal
// interval. For example, this could be an RDB that covers all of today.
//
// Metadata - Contains metadata on available functions in DAP (optional).
//
// Schema - Contains table/schema info (optional).
//
// Partitions - Partitions in the DAP.
//
purview:`ver`region`assetClass`startTS`endTS!(1;`amer;`equity;"p"$.z.D;0Wp)

params:(0#`)!()
params[`myAPI]:(
    `name`type`description`isReq!(`x;-9 9h;"param 1";1b);
    `name`type`description`isReq`default!(`y;10h;"param 2";0b;""));

// etc...

metadata:flip
    (`fn    ,`custom    ,`description   ,`params        ,`return)!flip(
    (`myAPI ;1b         ;"My API."      ;params`myAPI   ;`type`description!(98h;"Return value"))
    // etc...

columns:(0#`)!();
columns[`trade]:flip
    (`column    ,`typ   ,`attrMem   ,`attrIDisk ,`attrDisk  ,`isSerialized  ,`fk)!flip(
    (`sym       ;11h    ;`          ;`g         ;`p         ;0b             ;`instrument.sym);
    (`time      ;12h    ;`          ;`          ;`          ;0b             ;`);
    (`price     ;9h     ;`          ;`          ;`          ;0b             ;`);
    (`size      ;7h     ;`          ;`          ;`          ;0b             ;`);
    (`updTime   ;12h    ;`          ;`          ;`          ;0b             ;0b))

// etc...

schema:flip
    (`table ,`typ           ,`pkCols    ,`updTsCol  ,`sortColsMem   ,`sortColsIDisk ,`sortColsDisk  ,`columns)!flip(
    (`trade ;`partitioned   ;`sym`time  ;`updTime   ;`$()           ;enl`updTime    ;`sym`time      ;columns`trade);
    // etc...

partitions:enlist enlist[`date]!enlist .z.D // We contain data for only today, say

rc:hopen`:rcHost:1234 // Connect to the RC (alternatively, RC can be found via discovery)
neg[rc](`.sgrc.registerDAP;hsym`$string[.z.h],":",string system"p";1b;purview;`myAssembly;`RDB;metadata;schema;partitions) // Register with the RC

// Note that if we don't want to provide metadata/schema, we could register with the RC thusly:
//  neg[rc](`.sgrc.registerDAP;hsym`$string[.z.h],":",string system"p";1b;purview;`myAssembly;`RDB;();();partitions)


//
// Purview can be updated. For example, at the start of an EOD, a DAP may wish to
// announce itself as unavailable until the EOD ends. Upon EOD completion, it can
// roll its date (and its version number) forward and announce itself as available
// again.
//
onEodStart:{[]
    neg[rc](`.sgrc.updDapStatus;0b;()!();());
    }

onEodEnd:{[]
    purview[`ver]+:1; / Roll our version number forward
    purview[`startTS]+:1D; / Roll the date forward
    prtns:enlist enlist[`date]!enlist"d"purview`startTS; / Roll partition date forward
    neg[rc](`.sgrc.updDapStatus;1b;`ver`startTS#purview;prtns)
    }

//
// API execution. This function gets invoked by the GW on an API request. The DAP is
// obligated to respond to the Agg with its result, and the RC with an update that
// it available for more work. It should do this even in the event of an error.
//
// Note the name and signature of the function are important.
//
.da.execute:{[apiName;hdr;args]
    //
    // The hdr includes, among other things, `pvVer`, which is the version number
    // the RC used to make its selection. If the number is out of date, we should
    // NOT proceed as normal, since we may no longer cover the portion of the
    // request that the SG needs from us. We should error or trigger a retry
    // (Note: automatic retries NYI).
    //
    res:$[purview[`ver]=hdr`pvVer;
        @[apiName;args;`execErr]; / Run the API as normal
        `pvErr]; / Purview versions don't match

    //
    // The Agg and RC expect a response header with, at minimum:
    //  - the keys that, and
    //  - response and application codes (where 0 represents OK/no errors).
    //
    // Note: Error mapping in the Agg NYI.
    //
    respHdr:hdr,`rc`ac!
        $[res~`execErr;
            `rc`ac!10 10h; / General error code
        res~`pvErr;
            `rc`ac!13 30h; / Version mismatch error
        / OK
            `rc`ac!0 0h]; / OK

    //
    // Send response to the Agg. It is good practice to check whether the send was
    // successful. In the event that the send failed, we can notify the RC who
    // can complete the request on our behalf.
    //
    // Note, the Agg we are to report to is in hdr`agg.
    //
    sent:.[;(hdr`agg;respHdr;agg);0b]{
        h:hopen x; / Open connection to agg
        neg[h](`.sgagg.onPartial;y;z); / Send response
        hclose h; / Alternatively, cache this for future responses
        1b / Success
        }

    //
    // Notify the RC of completion so that it can select us for more work.
    // If the send to the Agg failed, we can use this opportunity to inform
    // the RC of the failure with the `sendErr` key in the response header.
    //
    if[not sent;respHdr[`sendErr]:1b]; / Include sendErr (note: value is arbitrary)
    neg[rc](`.sgrc.onPartial;respHdr); / Notify RC of completion
    }

//
// Implementation of APIs.
//
...