HPCC-18406 Use global and local ids to track related transactions

In order to create a giant graph of every system call related to a
global product order/request, an external system may pass a globalId
and a "callerId" to an HPCC component.

GlobalIds tie together the entire customer order/request across
all components.

CallerIds tie together all the calls made from a particular transaction
node.

When globalIds are recieved we should:

1. Log the global and caller ids that are received.
2. Create our own local id representing the specific request on
   the current node.
2. When making any SOAPCALL or HTTPCALL automatically pass along the
   globalId and pass our localId as the outgoing "callerId".

Signed-off-by: Anthony Fishbeck <anthony.fishbeck@lexisnexis.com>

common/thorhelper/thorsoapcall.cpp

@@ -1524,6 +1524,12 @@ private:
         if (!httpHeaderBlockContainsHeader(httpheaders, ACCEPT_ENCODING))
             request.appendf("%s: gzip, deflate\r\n", ACCEPT_ENCODING);
 #endif
+        if (!isEmptyString(master->logctx.queryGlobalId()))
+        {
+            request.append(master->logctx.queryGlobalIdHttpHeader()).append(": ").append(master->logctx.queryGlobalId()).append("\r\n");
+            if (!isEmptyString(master->logctx.queryLocalId()))
+                request.append(master->logctx.queryCallerIdHttpHeader()).append(": ").append(master->logctx.queryLocalId()).append("\r\n");  //our localId is reciever's callerId
+        }
 
         if (master->wscType == STsoap)
         {

ecl/eclagent/eclagent.cpp

@@ -2071,6 +2071,31 @@ void EclAgent::runProcess(IEclProcess *process)
     bool retainMemory = agentTopology->getPropBool("@heapRetainMemory", false);
     retainMemory = globals->getPropBool("heapRetainMemory", retainMemory);
 
+    if (globals->hasProp("@httpGlobalIdHeader"))
+        updateDummyContextLogger().setHttpIdHeaders(globals->queryProp("@httpGlobalIdHeader"), globals->queryProp("@httpCallerIdHeader"));
+
+    if (queryWorkUnit()->hasDebugValue("GlobalId"))
+    {
+        SCMStringBuffer globalId;
+        queryWorkUnit()->getDebugValue("GlobalId", globalId);
+        if (globalId.length())
+        {
+            SocketEndpoint thorEp;
+            thorEp.setLocalHost(0);
+            updateDummyContextLogger().setGlobalId(globalId.str(), thorEp, GetCurrentProcessId());
+
+            VStringBuffer msg("GlobalId: %s", globalId.str());
+            SCMStringBuffer txId;
+            queryWorkUnit()->getDebugValue("CallerId", txId);
+            if (txId.length())
+                msg.append(", CallerId: ").append(txId.str());
+            txId.set(updateDummyContextLogger().queryLocalId());
+            if (txId.length())
+                msg.append(", LocalId: ").append(txId.str());
+            updateDummyContextLogger().CTXLOG("%s", msg.str());
+        }
+    }
+
 #ifndef __64BIT__
     if (memLimitMB > 4096)
     {

esp/services/ws_ecl/ws_ecl_service.cpp

@@ -232,6 +232,13 @@ bool CWsEclService::init(const char * name, const char * type, IPropertyTree * c
     else
         workunitTimeout = WAIT_FOREVER;
 
+    const char *headerName = serviceTree->queryProp("HttpGlobalIdHeader");
+    if (headerName && *headerName && !streq(headerName, "HPCC-Global-Id")) //default will be checked anyway
+        globalIdHttpHeader.set(headerName);
+    headerName = serviceTree->queryProp("HttpCallerIdHeader");
+    if (headerName && *headerName && !streq(headerName, "HPCC-Caller-Id")) //default will be checked anyway
+        callerIdHttpHeader.set(headerName);
+
     Owned<IPropertyTreeIterator> cfgTargets = serviceTree->getElements("Targets/Target");
     ForEach(*cfgTargets)
         targets.append(cfgTargets->query().queryProp(NULL));
@@ -1850,7 +1857,7 @@ int CWsEclBinding::getWsEcl2Form(CHttpRequest* request, CHttpResponse* response,
     return 0;
 }
 
-int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, IPropertyTree *reqTree, StringBuffer &out, unsigned flags, TextMarkupFormat fmt, const char *viewname, const char *xsltname)
+int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, IPropertyTree *reqTree, StringBuffer &out, unsigned flags, CHttpRequest *httpreq, TextMarkupFormat fmt, const char *viewname, const char *xsltname)
 {
     IConstWorkUnit *sourceWorkUnit = wsinfo.ensureWorkUnit();
 
@@ -1871,6 +1878,23 @@ int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinf
 
     StringAttr wuid(workunit->queryWuid());  // NB queryWuid() not valid after workunit,clear()
 
+    if (httpreq)
+    {
+        StringBuffer globalId, callerId;
+        wsecl->getHttpGlobalIdHeader(httpreq, globalId);
+        wsecl->getHttpCallerIdHeader(httpreq, callerId);
+        if (globalId.length())
+        {
+            workunit->setDebugValue("GlobalId", globalId.str(), true);
+
+            SocketEndpoint ep;
+            StringBuffer localId;
+            appendLocalId(localId, httpreq->getSocket()->getEndpoint(ep), 0);
+            workunit->setDebugValue("CallerId", localId.str(), true); //our localId becomes caller id for the next hop
+            DBGLOG("GlobalId: %s, CallerId: %s, LocalId: %s, Wuid: %s", globalId.str(), callerId.str(), localId.str(), wuid.str());
+        }
+    }
+
     SCMStringBuffer token;
     createToken(wuid.str(), context.queryUserId(), context.queryPassword(), token);
     workunit->setSecurityToken(token.str());
@@ -1923,13 +1947,13 @@ int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinf
     return true;
 }
 
-int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, const char *xml, StringBuffer &out, unsigned flags, TextMarkupFormat fmt, const char *viewname, const char *xsltname)
+int CWsEclBinding::submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, const char *xml, StringBuffer &out, unsigned flags, CHttpRequest *httpreq, TextMarkupFormat fmt, const char *viewname, const char *xsltname)
 {
     Owned<IPropertyTree> reqTree = createPTreeFromXMLString(xml, ipt_ordered, (PTreeReaderOptions)(ptr_ignoreWhiteSpace|ptr_ignoreNameSpaces));
-    return submitWsEclWorkunit(context, wsinfo, reqTree, out, flags, fmt, viewname, xsltname);
+    return submitWsEclWorkunit(context, wsinfo, reqTree, out, flags, httpreq, fmt, viewname, xsltname);
 }
 
-void CWsEclBinding::sendRoxieRequest(const char *target, StringBuffer &req, StringBuffer &resp, StringBuffer &status, const char *query, bool trim, const char *contentType)
+void CWsEclBinding::sendRoxieRequest(const char *target, StringBuffer &req, StringBuffer &resp, StringBuffer &status, const char *query, bool trim, const char *contentType, CHttpRequest *httpreq)
 {
     ISmartSocketFactory *conn = NULL;
     SocketEndpoint ep;
@@ -1949,8 +1973,28 @@ void CWsEclBinding::sendRoxieRequest(const char *target, StringBuffer &req, Stri
         if (!trim)
             url.append("?.trim=0");
 
+        Owned<IProperties> headers;
         Owned<IHttpClient> httpclient = httpctx->createHttpClient(NULL, url);
         httpclient->setTimeOut(wsecl->roxieTimeout);
+        if (httpreq)
+        {
+            StringBuffer globalId, callerId;
+            wsecl->getHttpGlobalIdHeader(httpreq, globalId);
+            wsecl->getHttpCallerIdHeader(httpreq, callerId);
+
+            if (globalId.length())
+            {
+                headers.setown(createProperties());
+                headers->setProp(wsecl->queryGlobalIdHeaderName(), globalId);
+
+                SocketEndpoint ep;
+                StringBuffer localId;
+                appendLocalId(localId, httpreq->getSocket()->getEndpoint(ep), 0);
+                if (localId.length())
+                    headers->setProp(wsecl->queryCallerIdHeaderName(), localId);
+                DBGLOG("GlobalId: %s, CallerId: %s, LocalId: %s", globalId.str(), callerId.str(), localId.str());
+            }
+        }
         if (0 > httpclient->sendRequest("POST", contentType, req, resp, status))
             throw MakeStringException(-1, "Roxie cluster communication error: %s", target);
     }
@@ -2000,7 +2044,7 @@ int CWsEclBinding::onSubmitQueryOutput(IEspContext &context, CHttpRequest* reque
         getWsEclJsonRequest(jsonmsg, context, request, wsinfo, "json", NULL, REQSF_TRIM, false);
         if (jsonp && *jsonp)
             output.append(jsonp).append('(');
-        sendRoxieRequest(wsinfo.qsetname.get(), jsonmsg, output, status, wsinfo.queryname, trim, "application/json");
+        sendRoxieRequest(wsinfo.qsetname.get(), jsonmsg, output, status, wsinfo.queryname, trim, "application/json", request);
         if (jsonp && *jsonp)
             output.append(");");
     }
@@ -2017,11 +2061,11 @@ int CWsEclBinding::onSubmitQueryOutput(IEspContext &context, CHttpRequest* reque
         if (!format || !streq(format, "expanded"))
             xmlflags |= WWV_OMIT_SCHEMAS;
         if (!isRoxieReq)
-            submitWsEclWorkunit(context, wsinfo, soapmsg.str(), output, xmlflags, outputJSON ? MarkupFmt_JSON : MarkupFmt_XML);
+            submitWsEclWorkunit(context, wsinfo, soapmsg.str(), output, xmlflags, request, outputJSON ? MarkupFmt_JSON : MarkupFmt_XML);
         else
         {
             StringBuffer roxieresp;
-            sendRoxieRequest(wsinfo.qsetname, soapmsg, roxieresp, status, wsinfo.queryname, trim, "text/xml");
+            sendRoxieRequest(wsinfo.qsetname, soapmsg, roxieresp, status, wsinfo.queryname, trim, "text/xml", request);
             if (xmlflags & WWV_OMIT_SCHEMAS)
                 expandWuXmlResults(output, wsinfo.queryname, roxieresp.str(), xmlflags);
             else
@@ -2069,7 +2113,7 @@ int CWsEclBinding::onSubmitQueryOutputView(IEspContext &context, CHttpRequest* r
     const char *view = context.queryRequestParameters()->queryProp("view");
     if (strieq(clustertype.str(), "roxie"))
     {
-        sendRoxieRequest(wsinfo.qsetname.get(), soapmsg, output, status, wsinfo.queryname, false, "text/xml");
+        sendRoxieRequest(wsinfo.qsetname.get(), soapmsg, output, status, wsinfo.queryname, false, "text/xml", request);
         Owned<IWuWebView> web = createWuWebView(*wu, wsinfo.qsetname.get(), wsinfo.queryname.get(), getCFD(), true, queryXsltConfig());
         if (!view)
             web->applyResultsXSLT(xsltfile.str(), output.str(), html);
@@ -2078,7 +2122,7 @@ int CWsEclBinding::onSubmitQueryOutputView(IEspContext &context, CHttpRequest* r
     }
     else
     {
-        submitWsEclWorkunit(context, wsinfo, soapmsg.str(), html, 0, MarkupFmt_XML, view, xsltfile.str());
+        submitWsEclWorkunit(context, wsinfo, soapmsg.str(), html, 0, request, MarkupFmt_XML, view, xsltfile.str());
     }
 
     response->setContent(html.str());
@@ -2484,7 +2528,7 @@ int CWsEclBinding::onGet(CHttpRequest* request, CHttpResponse* response)
             StringBuffer status;
             if (getEspLogLevel()>LogNormal)
                 DBGLOG("roxie req: %s", soapreq.str());
-            sendRoxieRequest(target, soapreq, output, status, qid, parms->getPropBool(".trim", true), "text/xml");
+            sendRoxieRequest(target, soapreq, output, status, qid, parms->getPropBool(".trim", true), "text/xml", request);
             if (getEspLogLevel()>LogNormal)
                 DBGLOG("roxie resp: %s", output.str());
 
@@ -2672,7 +2716,7 @@ void CWsEclBinding::handleJSONPost(CHttpRequest *request, CHttpResponse *respons
 
         StringBuffer status;
         if (wsecl->connMap.getValue(queryset.str()))
-            sendRoxieRequest(queryset.str(), content, jsonresp, status, queryname.str(), trim, "application/json");
+            sendRoxieRequest(queryset.str(), content, jsonresp, status, queryname.str(), trim, "application/json", request);
         else
         {
             WsEclWuInfo wsinfo(wuid.str(), queryset.str(), queryname.str(), ctx->queryUserId(), ctx->queryPassword());
@@ -2691,7 +2735,7 @@ void CWsEclBinding::handleJSONPost(CHttpRequest *request, CHttpResponse *respons
                     break;
                 }
             }
-            submitWsEclWorkunit(*ctx, wsinfo, reqTree, jsonresp, 0, MarkupFmt_JSON);
+            submitWsEclWorkunit(*ctx, wsinfo, reqTree, jsonresp, 0, request, MarkupFmt_JSON);
         }
         if (jsonp && *jsonp)
             jsonresp.append(");");
@@ -2789,7 +2833,7 @@ int CWsEclBinding::HandleSoapRequest(CHttpRequest* request, CHttpResponse* respo
         bool trim = ctx->queryRequestParameters()->getPropBool(".trim", true);
         StringBuffer content(request->queryContent());
         StringBuffer output;
-        sendRoxieRequest(target, content, output, status, queryname, trim, "text/xml");
+        sendRoxieRequest(target, content, output, status, queryname, trim, "text/xml", request);
         if (!(xmlflags  & WWV_CDATA_SCHEMAS))
             soapresp.swapWith(output);
         else
@@ -2808,7 +2852,7 @@ int CWsEclBinding::HandleSoapRequest(CHttpRequest* request, CHttpResponse* respo
     else
     {
         WsEclWuInfo wsinfo(wuid.str(), target.str(), queryname.str(), ctx->queryUserId(), ctx->queryPassword());
-        submitWsEclWorkunit(*ctx, wsinfo, content.str(), soapresp, xmlflags);
+        submitWsEclWorkunit(*ctx, wsinfo, content.str(), soapresp, xmlflags, request);
     }
 
     if (getEspLogLevel()>LogNormal)

esp/services/ws_ecl/ws_ecl_service.hpp

@@ -107,6 +107,8 @@ public:
     StringArray targets;
     StringAttr auth_method;
     StringAttr portal_URL;
+    StringAttr globalIdHttpHeader;
+    StringAttr callerIdHttpHeader;
     unsigned roxieTimeout;
     unsigned workunitTimeout;
 
@@ -122,6 +124,34 @@ public:
     virtual bool init(const char * name, const char * type, IPropertyTree * cfg, const char * process);
     virtual void setContainer(IEspContainer * container){}
 
+    StringBuffer &getHttpGlobalIdHeader(CHttpRequest *request, StringBuffer &value)
+    {
+        if (!globalIdHttpHeader.isEmpty())
+            request->getHeader(globalIdHttpHeader, value);
+        if (value.isEmpty())
+            request->getHeader("HPCC-Global-Id", value); //always support receiving HPCC default
+        return value;
+    }
+    StringBuffer &getHttpCallerIdHeader(CHttpRequest *request, StringBuffer &value)
+    {
+        if (!callerIdHttpHeader.isEmpty())
+            request->getHeader(callerIdHttpHeader, value);
+        if (value.isEmpty())
+            request->getHeader("HPCC-Caller-Id", value); //always support receiving HPCC default
+        return value;
+    }
+    const char *queryGlobalIdHeaderName()
+    {
+        if (!globalIdHttpHeader.isEmpty())
+            return globalIdHttpHeader;
+        return "HPCC-Global-Id"; //HPCC default
+    }
+    const char *queryCallerIdHeaderName()
+    {
+        if (!callerIdHttpHeader.isEmpty())
+            return callerIdHttpHeader;
+        return "HPCC-Caller-Id"; //HPCC default
+    }
 };
 
 class CWsEclBinding : public CHttpSoapBinding
@@ -188,8 +218,8 @@ public:
     int onSubmitQueryOutput(IEspContext &context, CHttpRequest* request, CHttpResponse* response,    WsEclWuInfo &wsinfo, const char *format);
     int onSubmitQueryOutputView(IEspContext &context, CHttpRequest* request, CHttpResponse* response, WsEclWuInfo &wsinfo);
 
-    int submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, IPropertyTree *reqTree, StringBuffer &out, unsigned flags, TextMarkupFormat fmt=MarkupFmt_XML, const char *viewname=NULL, const char *xsltname=NULL);
-    int submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, const char *xml, StringBuffer &out, unsigned flags, TextMarkupFormat fmt=MarkupFmt_XML, const char *viewname=NULL, const char *xsltname=NULL);
+    int submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, IPropertyTree *reqTree, StringBuffer &out, unsigned flags, CHttpRequest *httpreq, TextMarkupFormat fmt=MarkupFmt_XML, const char *viewname=NULL, const char *xsltname=NULL);
+    int submitWsEclWorkunit(IEspContext & context, WsEclWuInfo &wsinfo, const char *xml, StringBuffer &out, unsigned flags, CHttpRequest *httpreq, TextMarkupFormat fmt=MarkupFmt_XML, const char *viewname=NULL, const char *xsltname=NULL);
 
     void handleHttpPost(CHttpRequest *request, CHttpResponse *response);
     void handleJSONPost(CHttpRequest *request, CHttpResponse *response);
@@ -213,7 +243,7 @@ public:
     void getWsEclJsonRequest(StringBuffer& soapmsg, IEspContext &context, CHttpRequest* request, WsEclWuInfo &wsinfo, const char *xmltype, const char *ns, unsigned flags, bool validate);
     void buildSampleResponseJSON(StringBuffer& msg, IEspContext &context, CHttpRequest* request, WsEclWuInfo &wsinfo);
 
-    void sendRoxieRequest(const char *process, StringBuffer &req, StringBuffer &resp, StringBuffer &status, const char *query, bool trim, const char *contentType);
+    void sendRoxieRequest(const char *process, StringBuffer &req, StringBuffer &resp, StringBuffer &status, const char *query, bool trim, const char *contentType, CHttpRequest *httpreq);
 };
 
 #endif //_WS_ECL_SERVICE_HPP__

initfiles/componentfiles/configxml/@temp/esp_service.xsl

@@ -906,6 +906,12 @@ xmlns:seisint="http://seisint.com"  xmlns:set="http://exslt.org/sets" exclude-re
                 <xsl:if test="string(@workunitTimout)!=''">
                     <WorkunitTimeout><xsl:value-of select="@workunitTimeout"/></WorkunitTimeout>
                 </xsl:if>
+                <xsl:if test="string(@httpCallerIdHeader)!=''">
+                    <HttpCallerIdHeader><xsl:value-of select="@httpCallerIdHeader"/></HttpCallerIdHeader>
+                </xsl:if>
+                <xsl:if test="string(@httpGlobalIdHeader)!=''">
+                    <HttpGlobalIdHeader><xsl:value-of select="@httpGlobalIdHeader"/></HttpGlobalIdHeader>
+                </xsl:if>
                 <WorkunitTimeout><xsl:value-of select="@workunitTimeout"/></WorkunitTimeout>
                 <xsl:if test="string(@xsltMaxDepth)!=''">
                     <xsltMaxDepth><xsl:value-of select="@xsltMaxDepth"/></xsltMaxDepth>

initfiles/componentfiles/configxml/agentexec.xsl

@@ -99,7 +99,14 @@
       <xsl:attribute name="thorConnectTimeout">
         <xsl:value-of select="@thorConnectTimeout"/>
       </xsl:attribute>
-
+      <xsl:if test="@httpGlobalIdHeader">
+        <xsl:attribute name="httpCallerIdHeader">
+          <xsl:value-of select="@httpCallerIdHeader"/>
+        </xsl:attribute>
+        <xsl:attribute name="httpGlobalIdHeader">
+          <xsl:value-of select="@httpGlobalIdHeader"/>
+        </xsl:attribute>
+      </xsl:if>
       <xsl:copy-of select="/Environment/Software/Directories"/>  
 
     </agentexec>

initfiles/componentfiles/configxml/eclagent_config.xsd.in

@@ -23,11 +23,11 @@
     <xs:element name="EclAgentProcess">
         <xs:complexType>
          <!--DOC-Autobuild-code-->
-	<xs:annotation>
-	   <xs:appinfo>
-	      <docid>EA.t1</docid>
-	   </xs:appinfo>
-	</xs:annotation>  
+       <xs:annotation>
+         <xs:appinfo>
+           <docid>EA.t1</docid>
+         </xs:appinfo>
+       </xs:annotation>
             <xs:sequence>
                 <xs:element name="Instance" maxOccurs="unbounded">
                     <xs:annotation>
@@ -248,6 +248,20 @@
         </xs:appinfo>
       </xs:annotation>
     </xs:attribute>
+    <xs:attribute name="httpCallerIdHeader" type="xs:string" use="optional" default="HPCC-Caller-Id">
+      <xs:annotation>
+        <xs:appinfo>
+          <tooltip>HTTP Header field to use for sending and receiving CallerId</tooltip>
+        </xs:appinfo>
+      </xs:annotation>
+    </xs:attribute>
+    <xs:attribute name="httpGlobalIdHeader" type="xs:string" use="optional" default="HPCC-Global-Id">
+      <xs:annotation>
+        <xs:appinfo>
+          <tooltip>HTTP Header field to use for sending and receiving GlobalId</tooltip>
+        </xs:appinfo>
+      </xs:annotation>
+    </xs:attribute>
   </xs:attributeGroup>
 
 </xs:schema>

initfiles/componentfiles/configxml/esp_service_wsecl2.xsd

@@ -31,11 +31,11 @@
                 </xs:annotation>
                 <xs:complexType>
                 <!--DocAutoBuildCode-->
-                    <xs:annotation>
-		       <xs:appinfo>
-		          <docid>MyWS2-T01</docid>
-		       </xs:appinfo>
-		    </xs:annotation>
+                  <xs:annotation>
+                    <xs:appinfo>
+                      <docid>MyWS2-T01</docid>
+                   </xs:appinfo>
+                  </xs:annotation>
            <xs:attribute name="name" type="xs:string" use="required" default="">
                     <xs:annotation>
                       <xs:appinfo>
@@ -93,26 +93,26 @@
                 <xs:complexType>
                 <!--DocAutoBuildCode-->
                  <xs:annotation>
-	           <xs:appinfo>
-                    <docid>MyWS2-T02</docid>
-	           </xs:appinfo>
+                   <xs:appinfo>
+                     <docid>MyWS2-T02</docid>
+                   </xs:appinfo>
                  </xs:annotation>
                   <xs:attribute name="name" type="topologyClusterType" use="required">
-	                <xs:annotation>
-	                  <xs:appinfo>
-	                    <tooltip>WsEcl will only display specified targets, if none specified WsEcl will display all targets.</tooltip>
-	                  </xs:appinfo>
-	                </xs:annotation>
+                    <xs:annotation>
+                      <xs:appinfo>
+                        <tooltip>WsEcl will only display specified targets, if none specified WsEcl will display all targets.</tooltip>
+                      </xs:appinfo>
+                    </xs:annotation>
                   </xs:attribute>
                 </xs:complexType>
               </xs:element>
             </xs:sequence>
             <!--DocAutoBuildCode-->
-                <xs:annotation>
-	          <xs:appinfo>
-                    <docid>MyWS2-T03</docid>
-	          </xs:appinfo>
-	        </xs:annotation>
+            <xs:annotation>
+              <xs:appinfo>
+                <docid>MyWS2-T03</docid>
+              </xs:appinfo>
+            </xs:annotation>
             <xs:attribute name="build" type="buildType" use="required">
                 <xs:annotation>
                     <xs:appinfo>
@@ -171,6 +171,20 @@
                     </xs:appinfo>
                 </xs:annotation>
             </xs:attribute>
+            <xs:attribute name="httpCallerIdHeader" type="xs:string" use="optional" default="HPCC-Caller-Id">
+              <xs:annotation>
+                <xs:appinfo>
+                  <tooltip>HTTP Header field to use for sending and receiving CallerId</tooltip>
+                </xs:appinfo>
+              </xs:annotation>
+            </xs:attribute>
+            <xs:attribute name="httpGlobalIdHeader" type="xs:string" use="optional" default="HPCC-Global-Id">
+              <xs:annotation>
+                <xs:appinfo>
+                  <tooltip>HTTP Header field to use for sending and receiving GlobalId</tooltip>
+                </xs:appinfo>
+              </xs:annotation>
+            </xs:attribute>
         </xs:complexType>
     </xs:element>
 </xs:schema>

initfiles/componentfiles/configxml/roxie.xsd.in

@@ -652,6 +652,20 @@
         </xs:appinfo>
       </xs:annotation>
     </xs:attribute>
+    <xs:attribute name="httpCallerIdHeader" type="xs:string" use="optional" default="HPCC-Caller-Id">
+      <xs:annotation>
+        <xs:appinfo>
+          <tooltip>HTTP Header field to use for sending and receiving CallerId</tooltip>
+        </xs:appinfo>
+      </xs:annotation>
+    </xs:attribute>
+    <xs:attribute name="httpGlobalIdHeader" type="xs:string" use="optional" default="HPCC-Global-Id">
+      <xs:annotation>
+        <xs:appinfo>
+          <tooltip>HTTP Header field to use for sending and receiving GlobalId</tooltip>
+        </xs:appinfo>
+      </xs:annotation>
+    </xs:attribute>
     <xs:attribute name="ignoreOrphans" type="xs:boolean" use="optional" default="true">
       <xs:annotation>
         <xs:appinfo>

initfiles/componentfiles/configxml/thor.xsd.in

@@ -641,6 +641,20 @@
           </xs:appinfo>
         </xs:annotation>
       </xs:attribute>
+      <xs:attribute name="httpCallerIdHeader" type="xs:string" use="optional" default="HPCC-Caller-Id">
+        <xs:annotation>
+          <xs:appinfo>
+            <tooltip>HTTP Header field to use for sending and receiving CallerId</tooltip>
+          </xs:appinfo>
+        </xs:annotation>
+      </xs:attribute>
+      <xs:attribute name="httpGlobalIdHeader" type="xs:string" use="optional" default="HPCC-Global-Id">
+        <xs:annotation>
+          <xs:appinfo>
+            <tooltip>HTTP Header field to use for sending and receiving GlobalId</tooltip>
+          </xs:appinfo>
+        </xs:annotation>
+      </xs:attribute>
     </xs:complexType>
     <xs:key name="thorProcessKey1">
       <xs:selector xpath="./ThorMasterProcess|./ThorSlaveProcess"/>

roxie/ccd/CMakeLists.txt

@@ -90,6 +90,7 @@ include_directories (
          ${HPCC_SOURCE_DIR}/dali/ft
          ${HPCC_SOURCE_DIR}/system/security/shared
          ${HPCC_SOURCE_DIR}/system/security/securesocket
+         ${HPCC_SOURCE_DIR}/system/libbase58
     )
 
 ADD_DEFINITIONS( -D_USRDLL -DCCD_EXPORTS -DSTARTQUERY_EXPORTS )
@@ -117,6 +118,7 @@ target_link_libraries ( ccd
          schedulectrl
          dllserver 
          workunit 
+         libbase58
     )
 
 IF (USE_OPENSSL)

roxie/ccd/ccd.hpp

@@ -568,6 +568,10 @@ public: // Not very clean but I don't care
     mutable bool aborted;
     mutable CIArrayOf<LogItem> log;
 private:
+    StringAttr globalIdHeader = "HPCC-Global-Id";
+    StringAttr callerIdHeader = "HPCC-Caller-Id";
+    StringAttr globalId;
+    StringBuffer localId;
     ContextLogger(const ContextLogger &);  // Disable copy constructor
 public:
     IMPLEMENT_IINTERFACE;
@@ -580,6 +584,8 @@ public:
         start = msTick();
         channel = 0;
         aborted = false;
+        if (topology->hasProp("@httpGlobalIdHeader"))
+            setHttpIdHeaders(topology->queryProp("@httpGlobalIdHeader"), topology->queryProp("@httpCallerIdHeader"));
     }
 
     void outputXML(IXmlStreamFlusher &out)
@@ -698,6 +704,34 @@ public:
     {
         stats.reset();
     }
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid)
+    {
+        globalId.set(id);
+        appendLocalId(localId.clear(), ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return globalId.get();
+    }
+    virtual const char *queryLocalId() const
+    {
+        return localId.str();
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        if (global && *global)
+            globalIdHeader.set(global);
+        if (caller && *caller)
+            callerIdHeader.set(caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return globalIdHeader.str();
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return callerIdHeader.str();
+    }
 };
 
 class StringContextLogger : public ContextLogger

roxie/ccd/ccdcontext.cpp

@@ -1161,6 +1161,7 @@ protected:
     mutable CriticalSection statsCrit;
     const IRoxieContextLogger &logctx;
 
+
 protected:
     bool exceptionLogged;
     bool aborted;
@@ -1310,6 +1311,31 @@ public:
     {
         return logctx.queryTraceLevel();
     }
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid)
+    {
+        const_cast<IRoxieContextLogger&>(logctx).setGlobalId(id, ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return logctx.queryGlobalId();
+    }
+    virtual const char *queryLocalId() const
+    {
+        return logctx.queryLocalId();
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        const_cast<IRoxieContextLogger&>(logctx).setHttpIdHeaders(global, caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return logctx.queryGlobalIdHttpHeader();
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return logctx.queryCallerIdHttpHeader();
+    }
+
 
     virtual void checkAbort()
     {

roxie/ccd/ccdlistener.cpp

@@ -1165,6 +1165,13 @@ public:
             wu.setown(daliHelper->attachWorkunit(wuid.get(), NULL));
         }
         Owned<StringContextLogger> logctx = new StringContextLogger(wuid.get());
+        if (wu->hasDebugValue("GlobalId"))
+        {
+            SCMStringBuffer globalId;
+            SocketEndpoint ep;
+            ep.setLocalHost(0);
+            logctx->setGlobalId(wu->getDebugValue("GlobalId", globalId).str(), ep, GetCurrentProcessId());
+        }
         Owned<IQueryFactory> queryFactory;
         try
         {
@@ -1280,7 +1287,23 @@ public:
         {
             StringBuffer s;
             logctx.getStats(s);
-            logctx.CTXLOG("COMPLETE: %s complete in %d msecs memory=%d Mb priority=%d slavesreply=%d%s", wuid.get(), elapsed, memused, priority, slavesReplyLen, s.str());
+
+            StringBuffer txidInfo;
+            const char *globalId = logctx.queryGlobalId();
+            if (globalId && *globalId)
+            {
+                txidInfo.append(" [GlobalId: ").append(globalId);
+                SCMStringBuffer s;
+                wu->getDebugValue("CallerId", s);
+                if (s.length())
+                    txidInfo.append(", CallerId: ").append(s.str());
+                s.set(logctx.queryLocalId());
+                if (s.length())
+                    txidInfo.append(", LocalId: ").append(s.str());
+                txidInfo.append(']');
+            }
+
+            logctx.CTXLOG("COMPLETE: %s%s complete in %d msecs memory=%d Mb priority=%d slavesreply=%d%s", wuid.get(), txidInfo.str(), elapsed, memused, priority, slavesReplyLen, s.str());
         }
     }
 
@@ -1311,6 +1334,7 @@ class RoxieProtocolMsgContext : implements IHpccProtocolMsgContext, public CInte
 public:
     StringAttr queryName;
     StringAttr uid = "-";
+    StringAttr callerId;
     Owned<CascadeManager> cascade;
     Owned<IDebuggerContext> debuggerContext;
     Owned<CDebugCommandHandler> debugCmdHandler;
@@ -1388,15 +1412,25 @@ public:
         return *cascade;
     }
 
-    virtual void setTransactionId(const char *id)
+    virtual void setTransactionId(const char *id, bool global)
     {
         if (!id || !*id)
             return;
         uid.set(id);
         ensureContextLogger();
+        if (!isEmptyString(logctx->queryGlobalId())) //globalId wins
+            return;
+        if (global)
+            logctx->setGlobalId(id, ep, 0);
         StringBuffer s;
         logctx->set(ep.getIpText(s).appendf(":%u{%s}", ep.port, uid.str()).str());
     }
+    virtual void setCallerId(const char *id)
+    {
+        if (!id || !*id)
+            return;
+        callerId.set(id);
+    }
     inline IDebuggerContext &ensureDebuggerContext(const char *id)
     {
         if (!debuggerContext)
@@ -1513,7 +1547,20 @@ public:
             {
                 StringBuffer s;
                 logctx->getStats(s);
-                logctx->CTXLOG("COMPLETE: %s %s from %s complete in %d msecs memory=%d Mb priority=%d slavesreply=%d resultsize=%d continue=%d%s", queryName.get(), uid.get(), peer, elapsed, memused, getQueryPriority(), slavesReplyLen, bytesOut, continuationNeeded, s.str());
+
+                StringBuffer txIds;
+                if (callerId.length())
+                    txIds.appendf("caller: %s", callerId.str());
+                const char *localId = logctx->queryLocalId();
+                if (localId && *localId)
+                {
+                    if (txIds.length())
+                        txIds.append(", ");
+                    txIds.append("local: ").append(localId);
+                }
+                if (txIds.length());
+                    txIds.insert(0, '[').append(']');
+                logctx->CTXLOG("COMPLETE: %s %s%s from %s complete in %d msecs memory=%d Mb priority=%d slavesreply=%d resultsize=%d continue=%d%s", queryName.get(), uid.get(), txIds.str(), peer, elapsed, memused, getQueryPriority(), slavesReplyLen, bytesOut, continuationNeeded, s.str());
             }
         }
     }

roxie/ccd/ccdprotocol.cpp

@@ -1756,6 +1756,16 @@ readAnother:
             {
                 mlResponseFmt = httpHelper.queryResponseMlFormat();
                 mlRequestFmt = httpHelper.queryRequestMlFormat();
+                const char *value = httpHelper.queryRequestHeader(logctx.queryGlobalIdHttpHeader());
+                if (!value || !*value)
+                    value = httpHelper.queryRequestHeader("HPCC-Global-Id"); //always support receiving in the HPCC form
+                if (value && *value)
+                    msgctx->setTransactionId(value, true);  //logged and forwarded through SOAPCALL/HTTPCALL
+                value = httpHelper.queryRequestHeader(logctx.queryCallerIdHttpHeader());
+                if (!value || !*value)
+                    value = httpHelper.queryRequestHeader("HPCC-Caller-Id");
+                if (value && *value)
+                    msgctx->setCallerId(value);  //only logged
             }
         }
 
@@ -1858,7 +1868,7 @@ readAnother:
                 uid = NULL;
                 sanitizeQuery(queryPT, queryName, sanitizedText, httpHelper, uid, isBlind, isDebug);
                 if (uid)
-                    msgctx->setTransactionId(uid);
+                    msgctx->setTransactionId(uid, false);
                 else
                     uid = "-";
 

roxie/ccd/ccdserver.cpp

@@ -249,6 +249,32 @@ public:
     {
         return ctx->isBlind();
     }
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid)
+    {
+        ctx->setGlobalId(id, ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return ctx->queryGlobalId();
+    }
+    virtual const char *queryLocalId() const
+    {
+        return ctx->queryLocalId();
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        ctx->setHttpIdHeaders(global, caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return ctx->queryGlobalIdHttpHeader();
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return ctx->queryCallerIdHttpHeader();
+    }
+
+
     virtual const QueryOptions &queryOptions() const
     {
         return ctx->queryOptions();
@@ -1148,6 +1174,33 @@ public:
             return traceLevel;
     }
 
+    virtual void setGlobalId(const char *id, SocketEndpoint&ep, unsigned pid)
+    {
+        if (ctx)
+            ctx->setGlobalId(id, ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return ctx ? ctx->queryGlobalId() : nullptr;
+    }
+    virtual const char *queryLocalId() const
+    {
+        return ctx ? ctx->queryLocalId() : nullptr;
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        if (ctx)
+            ctx->setHttpIdHeaders(global, caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return ctx ? ctx->queryGlobalIdHttpHeader() : "HPCC-Global-Id";
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return ctx ? ctx->queryCallerIdHttpHeader() : "HPCC-Caller-Id";
+    }
+
     virtual bool isPassThrough()
     {
         return false;

roxie/ccd/hpccprotocol.hpp

@@ -40,8 +40,9 @@ interface IHpccProtocolMsgContext : extends IInterface
     virtual void setIntercept(bool val) = 0;
     virtual bool getIntercept() = 0;
     virtual void outputLogXML(IXmlStreamFlusher &out) = 0;
-    virtual void setTransactionId(const char *id) = 0;
     virtual void writeLogXML(IXmlWriter &writer) = 0;
+    virtual void setTransactionId(const char *id, bool global) = 0;
+    virtual void setCallerId(const char *id) = 0;
 };
 
 interface IHpccProtocolResultsWriter : extends IInterface

system/CMakeLists.txt

@@ -19,6 +19,7 @@ HPCC_ADD_SUBDIRECTORY (jhtree)
 HPCC_ADD_SUBDIRECTORY (jlib)
 
 if (NOT JLIB_ONLY)
+   HPCC_ADD_SUBDIRECTORY (libbase58)
    HPCC_ADD_SUBDIRECTORY (hrpc)
    HPCC_ADD_SUBDIRECTORY (tbb_sm)
    HPCC_ADD_SUBDIRECTORY (mp)

system/jlib/CMakeLists.txt

@@ -180,6 +180,7 @@ include_directories (
          ../../system/include 
          ../../system/lzma
          ../../system/lz4_sm/lz4/lib
+         ../../system/libbase58
          ${CMAKE_CURRENT_BINARY_DIR}  # for generated jelog.h file 
          ${CMAKE_BINARY_DIR}
          ${CMAKE_BINARY_DIR}/oss
@@ -191,6 +192,7 @@ HPCC_ADD_LIBRARY( jlib SHARED ${SRCS} ${INCLUDES} )
 target_link_libraries ( jlib
         lzma
         lz4
+        libbase58
        )
 
 if ( ${HAVE_LIBDL} )

system/jlib/jlog.cpp

@@ -29,6 +29,8 @@
 #include "jmisc.hpp"
 #include "jprop.hpp"
 
+#include "libbase58.h"
+
 #define MSGCOMP_NUMBER 1000
 #define FILE_LOG_ENABLES_QUEUEUING
 
@@ -2550,6 +2552,12 @@ void IContextLogger::logOperatorException(IException *E, const char *file, unsig
 
 class DummyLogCtx : implements IContextLogger
 {
+private:
+    StringAttr globalId;
+    StringBuffer localId;
+    StringAttr globalIdHeader;
+    StringAttr callerIdHeader;
+
 public:
     // It's a static object - we don't want to actually link-count it...
     virtual void Link() const {}
@@ -2585,6 +2593,34 @@ public:
     {
         return 0;
     }
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid)
+    {
+        globalId.set(id);
+        appendLocalId(localId.clear(), ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return globalId.get();
+    }
+    virtual const char *queryLocalId() const
+    {
+        return localId.str();
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        if (global && *global)
+            globalIdHeader.set(global);
+        if (caller && *caller)
+            callerIdHeader.set(caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return globalIdHeader.str();
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return callerIdHeader.str();
+    }
 } dummyContextLogger;
 
 extern jlib_decl const IContextLogger &queryDummyContextLogger()
@@ -2592,6 +2628,34 @@ extern jlib_decl const IContextLogger &queryDummyContextLogger()
     return dummyContextLogger;
 }
 
+extern jlib_decl IContextLogger &updateDummyContextLogger()
+{
+    return dummyContextLogger;
+}
+
+extern jlib_decl StringBuffer &appendLocalId(StringBuffer &s, const SocketEndpoint &ep, unsigned pid)
+{
+    static unsigned short cnt = msTick();
+
+    MemoryBuffer data;
+    data.append(ep.iphash());
+    if (pid>0)
+        data.append(pid);
+    else
+        data.append(ep.port);
+    data.append(++cnt);
+
+    size_t b58Length = data.length() * 2;
+    StringBuffer id;
+
+    //base58 works well as a human readable format, i.e. so customers can easily report the id that was recieved
+    if (b58enc(id.reserve(b58Length), &b58Length, data.toByteArray(), data.length()) && b58Length > 1)
+    {
+        id.setLength(b58Length);
+        s.append(id);
+    }
+    return s;
+}
 
 extern jlib_decl void UseSysLogForOperatorMessages(bool use)
 {

system/jlib/jlog.hpp

@@ -1012,9 +1012,19 @@ interface jlib_decl IContextLogger : extends IInterface
     virtual void noteStatistic(StatisticKind kind, unsigned __int64 value) const = 0;
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const = 0;
     virtual unsigned queryTraceLevel() const = 0;
+
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid) = 0;
+    virtual void setHttpIdHeaders(const char *global, const char *caller) = 0;
+    virtual const char *queryGlobalId() const = 0;
+    virtual const char *queryLocalId() const = 0;
+    virtual const char *queryGlobalIdHttpHeader() const = 0;
+    virtual const char *queryCallerIdHttpHeader() const = 0;
 };
 
+extern jlib_decl StringBuffer &appendLocalId(StringBuffer &s, const SocketEndpoint &ep, unsigned pid);
+
 extern jlib_decl const IContextLogger &queryDummyContextLogger();
+extern jlib_decl IContextLogger &updateDummyContextLogger();
 
 //---------------------------------------------------------------------------
 

system/libbase58/AUTHORS

@@ -0,0 +1,2 @@
+Luke Dashjr <luke-jr+libbase58@utopios.org>
+Huang Le <4tarhl@gmail.com>

system/libbase58/CMakeLists.txt

@@ -0,0 +1,36 @@
+################################################################################
+#    HPCC SYSTEMS software Copyright (C) 2017 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+################################################################################
+
+# Component: libbase58 
+
+#####################################################
+# Description:
+# ------------
+#    Cmake Input File for libbase58
+#####################################################
+
+
+project( libbase58 ) 
+
+set ( SRCS
+        base58.c
+)
+
+ADD_DEFINITIONS( -D_LIB )
+
+SET_SOURCE_FILES_PROPERTIES( ${SRCS} PROPERTIES LANGUAGE CXX )
+HPCC_ADD_LIBRARY( libbase58 STATIC ${SRCS} )
+

system/libbase58/COPYING

@@ -0,0 +1,19 @@
+Copyright (c) 2014 Luke Dashjr
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

system/libbase58/README

@@ -0,0 +1,56 @@
+Initialisation
+--------------
+
+Before you can use libbase58 for base58check, you must provide a SHA256
+function. The required function signature is:
+	bool my_sha256(void *digest, const void *data, size_t datasz)
+Simply assign your function to b58_sha256_impl:
+	b58_sha256_impl = my_sha256;
+
+This is only required if base58check is used. Raw base58 does not need SHA256.
+
+
+Decoding Base58
+---------------
+
+Simply allocate a buffer to store the binary data in, and set a variable with
+the buffer size, and call the b58tobin function:
+	bool b58tobin(void *bin, size_t *binsz, const char *b58, size_t b58sz)
+The "canonical" base58 byte length will be assigned to binsz on success, which
+may be larger than the actual buffer if the input has many leading zeros.
+Regardless of the canonical byte length, the full binary buffer will be used.
+If b58sz is zero, it will be initialised with strlen(b58); note that a true
+zero-length base58 string is not supported here.
+
+
+Validating Base58Check
+----------------------
+
+After calling b58tobin, you can validate base58check data using the b58check
+function:
+	int b58check(const void *bin, size_t binsz, const char *b58, size_t b58sz)
+Call it with the same buffers used for b58tobin. If the return value is
+negative, an error occurred. Otherwise, the return value is the base58check
+"version" byte from the decoded data.
+
+
+Encoding Base58
+---------------
+
+Allocate a string to store the base58 content, create a size_t variable with the
+size of that allocation, and call:
+	bool b58enc(char *b58, size_t *b58sz, const void *data, size_t binsz)
+Note that you must pass a pointer to the string size variable, not the size
+itself. When b58enc returns, the variable will be modified to contain the actual
+number of bytes used (including the null terminator). If encoding fails for any
+reason, or if the string buffer is not large enough for the result, b58enc will
+return false. Otherwise, it returns true to indicate success.
+
+
+Encoding Base58Check
+--------------------
+
+Targeting base58check is done similarly to raw base58 encoding, but you must
+also provide a version byte:
+	bool b58check_enc(char *b58c, size_t *b58c_sz, uint8_t ver,
+	                  const void *data, size_t datasz)

system/libbase58/base58.c

@@ -0,0 +1,201 @@
+/*
+ * Copyright 2012-2014 Luke Dashjr
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the standard MIT license.  See COPYING for more details.
+ */
+
+#ifndef WIN32
+#include <arpa/inet.h>
+#else
+#include <winsock2.h>
+#endif
+
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <string.h>
+#include <sys/types.h>
+
+#include "libbase58.h"
+
+bool (*b58_sha256_impl)(void *, const void *, size_t) = NULL;
+
+static const int8_t b58digits_map[] = {
+    -1,-1,-1,-1,-1,-1,-1,-1, -1,-1,-1,-1,-1,-1,-1,-1,
+    -1,-1,-1,-1,-1,-1,-1,-1, -1,-1,-1,-1,-1,-1,-1,-1,
+    -1,-1,-1,-1,-1,-1,-1,-1, -1,-1,-1,-1,-1,-1,-1,-1,
+    -1, 0, 1, 2, 3, 4, 5, 6,  7, 8,-1,-1,-1,-1,-1,-1,
+    -1, 9,10,11,12,13,14,15, 16,-1,17,18,19,20,21,-1,
+    22,23,24,25,26,27,28,29, 30,31,32,-1,-1,-1,-1,-1,
+    -1,33,34,35,36,37,38,39, 40,41,42,43,-1,44,45,46,
+    47,48,49,50,51,52,53,54, 55,56,57,-1,-1,-1,-1,-1,
+};
+
+bool b58tobin(void *bin, size_t *binszp, const char *b58, size_t b58sz)
+{
+    size_t binsz = *binszp;
+    const unsigned char *b58u = (const unsigned char *)b58;
+    unsigned char *binu = (unsigned char *)bin;
+    size_t outisz = (binsz + 3) / 4;
+    uint32_t outi[outisz];
+    uint64_t t;
+    uint32_t c;
+    size_t i, j;
+    uint8_t bytesleft = binsz % 4;
+    uint32_t zeromask = bytesleft ? (0xffffffff << (bytesleft * 8)) : 0;
+    unsigned zerocount = 0;
+
+    if (!b58sz)
+        b58sz = strlen(b58);
+
+    memset(outi, 0, outisz * sizeof(*outi));
+
+    // Leading zeros, just count
+    for (i = 0; i < b58sz && b58u[i] == '1'; ++i)
+        ++zerocount;
+
+    for ( ; i < b58sz; ++i)
+    {
+        if (b58u[i] & 0x80)
+            // High-bit set on invalid digit
+            return false;
+        if (b58digits_map[b58u[i]] == -1)
+            // Invalid base58 digit
+            return false;
+        c = (unsigned)b58digits_map[b58u[i]];
+        for (j = outisz; j--; )
+        {
+            t = ((uint64_t)outi[j]) * 58 + c;
+            c = (t & 0x3f00000000) >> 32;
+            outi[j] = t & 0xffffffff;
+        }
+        if (c)
+            // Output number too big (carry to the next int32)
+            return false;
+        if (outi[0] & zeromask)
+            // Output number too big (last int32 filled too far)
+            return false;
+    }
+
+    j = 0;
+    switch (bytesleft) {
+        case 3:
+            *(binu++) = (outi[0] &   0xff0000) >> 16;
+        case 2:
+            *(binu++) = (outi[0] &     0xff00) >>  8;
+        case 1:
+            *(binu++) = (outi[0] &       0xff);
+            ++j;
+        default:
+            break;
+    }
+
+    for (; j < outisz; ++j)
+    {
+        *(binu++) = (outi[j] >> 0x18) & 0xff;
+        *(binu++) = (outi[j] >> 0x10) & 0xff;
+        *(binu++) = (outi[j] >>    8) & 0xff;
+        *(binu++) = (outi[j] >>    0) & 0xff;
+    }
+
+    // Count canonical base58 byte count
+    binu = (unsigned char *) bin;
+    for (i = 0; i < binsz; ++i)
+    {
+        if (binu[i])
+            break;
+        --*binszp;
+    }
+    *binszp += zerocount;
+
+    return true;
+}
+
+static
+bool my_dblsha256(void *hash, const void *data, size_t datasz)
+{
+    uint8_t buf[0x20];
+    return b58_sha256_impl(buf, data, datasz) && b58_sha256_impl(hash, buf, sizeof(buf));
+}
+
+int b58check(const void *bin, size_t binsz, const char *base58str, size_t b58sz)
+{
+    unsigned char buf[32];
+    const uint8_t *binc = (const uint8_t *) bin;
+    unsigned i;
+    if (binsz < 4)
+        return -4;
+    if (!my_dblsha256(buf, bin, binsz - 4))
+        return -2;
+    if (memcmp(&binc[binsz - 4], buf, 4))
+        return -1;
+
+    // Check number of zeros is correct AFTER verifying checksum (to avoid possibility of accessing base58str beyond the end)
+    for (i = 0; binc[i] == '\0' && base58str[i] == '1'; ++i)
+    {}  // Just finding the end of zeros, nothing to do in loop
+    if (binc[i] == '\0' || base58str[i] == '1')
+        return -3;
+
+    return binc[0];
+}
+
+static const char b58digits_ordered[] = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";
+
+bool b58enc(char *b58, size_t *b58sz, const void *data, size_t binsz)
+{
+    const uint8_t *bin = (const uint8_t *) data;
+    int carry;
+    ssize_t i, j, high, zcount = 0;
+    size_t size;
+
+    while (zcount < binsz && !bin[zcount])
+        ++zcount;
+
+    size = (binsz - zcount) * 138 / 100 + 1;
+    uint8_t buf[size];
+    memset(buf, 0, size);
+
+    for (i = zcount, high = size - 1; i < binsz; ++i, high = j)
+    {
+        for (carry = bin[i], j = size - 1; (j > high) || carry; --j)
+        {
+            carry += 256 * buf[j];
+            buf[j] = carry % 58;
+            carry /= 58;
+        }
+    }
+
+    for (j = 0; j < size && !buf[j]; ++j);
+
+    if (*b58sz <= zcount + size - j)
+    {
+        *b58sz = zcount + size - j + 1;
+        return false;
+    }
+
+    if (zcount)
+        memset(b58, '1', zcount);
+    for (i = zcount; j < size; ++i, ++j)
+        b58[i] = b58digits_ordered[buf[j]];
+    b58[i] = '\0';
+    *b58sz = i + 1;
+
+    return true;
+}
+
+bool b58check_enc(char *b58c, size_t *b58c_sz, uint8_t ver, const void *data, size_t datasz)
+{
+    uint8_t buf[1 + datasz + 0x20];
+    uint8_t *hash = &buf[1 + datasz];
+
+    buf[0] = ver;
+    memcpy(&buf[1], data, datasz);
+    if (!my_dblsha256(hash, buf, datasz + 1))
+    {
+        *b58c_sz = 0;
+        return false;
+    }
+
+    return b58enc(b58c, b58c_sz, buf, 1 + datasz + 4);
+}

system/libbase58/clitool.c

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2014 Luke Dashjr
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the standard MIT license.  See COPYING for more details.
+ */
+
+#include <ctype.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <gcrypt.h>
+
+#include "libbase58.h"
+
+static
+bool my_sha256(void *digest, const void *data, size_t datasz)
+{
+	gcry_md_hash_buffer(GCRY_MD_SHA256, digest, data, datasz);
+	return true;
+}
+
+static
+void usage(const char *prog)
+{
+	fprintf(stderr, "Usage: %s [-c] [-d] [data]\n", prog);
+	fprintf(stderr, "\t-c         Use base58check (default: raw base58)\n");
+	fprintf(stderr, "\t-d <size>  Decode <size> bytes\n");
+	exit(1);
+}
+
+int main(int argc, char **argv)
+{
+	bool b58c = false;
+	size_t decode = 0;
+	int opt;
+	while ( (opt = getopt(argc, argv, "cd:h")) != -1)
+	{
+		switch (opt)
+		{
+			case 'c':
+				b58c = true;
+				b58_sha256_impl = my_sha256;
+				break;
+			case 'd':
+			{
+				int i = atoi(optarg);
+				if (i < 0 || (uintmax_t)i >= SIZE_MAX)
+					usage(argv[0]);
+				decode = (size_t)i;
+				break;
+			}
+			default:
+				usage(argv[0]);
+		}
+	}
+	
+	size_t rt;
+	union {
+		uint8_t *b;
+		char *s;
+	} r;
+	if (optind >= argc)
+	{
+		rt = 0;
+		r.b = NULL;
+		while (!feof(stdin))
+		{
+			r.b = realloc(r.b, rt + 0x100);
+			rt += fread(&r.b[rt], 1, 0x100, stdin);
+		}
+		if (decode)
+			while (isspace(r.s[rt-1]))
+				--rt;
+	}
+	else
+	{
+		r.s = argv[optind];
+		rt = strlen(argv[optind]);
+	}
+	
+	if (decode)
+	{
+		uint8_t bin[decode];
+		size_t ssz = decode;
+		if (!b58tobin(bin, &ssz, r.s, rt))
+			return 2;
+		if (b58c)
+		{
+			int chk = b58check(bin, decode, r.s, rt);
+			if (chk < 0)
+				return chk;
+			if (fwrite(bin, decode, 1, stdout) != 1)
+				return 3;
+		}
+		else
+		{
+			// Raw base58 doesn't check length match
+			uint8_t cbin[ssz];
+			if (ssz > decode)
+			{
+				size_t zeros = ssz - decode;
+				memset(cbin, 0, zeros);
+				memcpy(&cbin[zeros], bin, decode);
+			}
+			else
+				memcpy(cbin, &bin[decode - ssz], ssz);
+			
+			if (fwrite(cbin, ssz, 1, stdout) != 1)
+				return 3;
+		}
+	}
+	else
+	{
+		size_t ssz = rt * 2;
+		char s[ssz];
+		bool rv;
+		if (b58c)
+			rv = rt && b58check_enc(s, &ssz, r.b[0], &r.b[1], rt-1);
+		else
+			rv = b58enc(s, &ssz, r.b, rt);
+		if (!rv)
+			return 2;
+		puts(s);
+	}
+}

system/libbase58/libbase58.h

@@ -0,0 +1,23 @@
+#ifndef LIBBASE58_H
+#define LIBBASE58_H
+
+#include <stdbool.h>
+#include <stddef.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern bool (*b58_sha256_impl)(void *, const void *, size_t);
+
+extern bool b58tobin(void *bin, size_t *binsz, const char *b58, size_t b58sz);
+extern int b58check(const void *bin, size_t binsz, const char *b58, size_t b58sz);
+
+extern bool b58enc(char *b58, size_t *b58sz, const void *bin, size_t binsz);
+extern bool b58check_enc(char *b58c, size_t *b58c_sz, uint8_t ver, const void *data, size_t datasz);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

system/libbase58/tests/decode-b58c-fail.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+! base58 -d 25 -c 19DXstMaV43WpYg4ceREiiTv2UntmoiA9a >/dev/null

system/libbase58/tests/decode-b58c-null.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(base58 -d 25 -c 19DXstMaV43WpYg4ceREiiTv2UntmoiA9a | xxd -p)
+test x$hex = x

system/libbase58/tests/decode-b58c-toolong.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+! base58 -d 25 -c 1119DXstMaV43WpYg4ceREiiTv2UntmoiA9a >/dev/null

system/libbase58/tests/decode-b58c-tooshort.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+! base58 -d 25 -c 111111111111111111114oLvT2 >/dev/null

system/libbase58/tests/decode-b58c.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(base58 -d 25 -c 19DXstMaV43WpYg4ceREiiTv2UntmoiA9j | xxd -p)
+test x$hex != x005a1fc5dd9e6f03819fca94a2d89669469667f9a1

system/libbase58/tests/decode-highbit-prefix.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(echo 993233 | xxd -r -p | base58 -d 25 || echo FAIL)
+test "x${hex}" = "xFAIL"

system/libbase58/tests/decode-highbit.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(echo 319932 | xxd -r -p | base58 -d 25 || echo FAIL)
+test "x${hex}" = "xFAIL"

system/libbase58/tests/decode-small.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(base58 -d 4 2 | xxd -p)
+test x$hex = x01

system/libbase58/tests/decode-zero.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(base58 -d 25 111111 | xxd -p)
+test x$hex = x000000000000

system/libbase58/tests/decode.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+hex=$(base58 -d 50 19DXstMaV43WpYg4ceREiiTv2UntmoiA9j | xxd -p)
+test x$hex = x005a1fc5dd9e6f03819fca94a2d89669469667f9a074655946

system/libbase58/tests/encode-b58c.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+b58=$(echo '005a1fc5dd9e6f03819fca94a2d89669469667f9a0' | xxd -r -p | base58 -c)
+test x$b58 = x19DXstMaV43WpYg4ceREiiTv2UntmoiA9j

system/libbase58/tests/encode-fail.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+b58=$(echo '005a1fc5dd9e6f03819fca94a2d89669469667f9a174655946' | xxd -r -p | base58)
+test x$b58 != x19DXstMaV43WpYg4ceREiiTv2UntmoiA9j

system/libbase58/tests/encode-neg-index.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+# This input causes the loop iteration counter to go negative
+b58=$(echo '00CEF022FA' | xxd -r -p | base58)
+test x$b58 = x16Ho7Hs

system/libbase58/tests/encode-small.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+b58=$(base58 1)
+test x$b58 = xr

system/libbase58/tests/encode.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+b58=$(echo '005a1fc5dd9e6f03819fca94a2d89669469667f9a074655946' | xxd -r -p | base58)
+test x$b58 = x19DXstMaV43WpYg4ceREiiTv2UntmoiA9j

testing/regress/ecl/httpcall_multiheader.ecl

@@ -15,13 +15,13 @@ string TargetURL := 'http://' + TargetIP + ':8010/WsSmc/HttpEcho?name=doe,joe&nu
 
 string constHeader := 'constHeaderValue';
 
-httpcallResult := HTTPCALL(TargetURL,'GET', 'text/xml', httpEchoServiceResponseRecord, xpath('Envelope/Body/HttpEchoResponse'),httpheader('literalHeader','literalValue'), httpheader('constHeader','constHeaderValue'), httpheader('storedHeader', storedHeader));
+httpcallResult := HTTPCALL(TargetURL,'GET', 'text/xml', httpEchoServiceResponseRecord, xpath('Envelope/Body/HttpEchoResponse'),httpheader('literalHeader','literalValue'), httpheader('constHeader','constHeaderValue'), httpheader('storedHeader', storedHeader), httpheader('HPCC-Global-Id','9876543210'), httpheader('HPCC-Caller-Id','http111'));
 output(httpcallResult, named('httpcallResult'));
 
 //test proxyaddress functionality by using an invalid targetUrl, but a valid proxyaddress.  HTTP Host header will be wrong, but should still work fine as it's ignored by ESP.
 string hostURL := 'http://1.1.1.1:9999/WsSmc/HttpEcho?name=doe,joe&number=1';
 string targetProxy := 'http://' + TargetIP + ':8010';
 
-proxyResult := HTTPCALL(hostURL,'GET', 'text/xml', httpEchoServiceResponseRecord, xpath('Envelope/Body/HttpEchoResponse'), proxyaddress(targetProxy), httpheader('literalHeader','literalValue'), httpheader('constHeader','constHeaderValue'), httpheader('storedHeader', storedHeader));
+proxyResult := HTTPCALL(hostURL,'GET', 'text/xml', httpEchoServiceResponseRecord, xpath('Envelope/Body/HttpEchoResponse'), proxyaddress(targetProxy), httpheader('literalHeader','literalValue'), httpheader('constHeader','constHeaderValue'), httpheader('storedHeader', storedHeader), httpheader('HPCC-Global-Id','9876543210'), httpheader('HPCC-Caller-Id','http222'));
 
 output(proxyResult, named('proxyResult'));

testing/regress/ecl/key/httpcall_multiheader.xml

@@ -1,6 +1,6 @@
 <Dataset name='httpcallResult'>
- <Row><Method>GET</Method><UrlPath>/WsSmc/HttpEcho</UrlPath><UrlParameters>name=doe,joe&amp;number=1</UrlParameters><Headers><Header>Accept-Encoding: gzip, deflate</Header><Header>Accept: text/xml</Header><Header>constHeader: constHeaderValue</Header><Header>literalHeader: literalValue</Header><Header>storedHeader: StoredHeaderDefault</Header></Headers><Content></Content></Row>
+ <Row><Method>GET</Method><UrlPath>/WsSmc/HttpEcho</UrlPath><UrlParameters>name=doe,joe&amp;number=1</UrlParameters><Headers><Header>Accept-Encoding: gzip, deflate</Header><Header>Accept: text/xml</Header><Header>HPCC-Caller-Id: http111</Header><Header>HPCC-Global-Id: 9876543210</Header><Header>constHeader: constHeaderValue</Header><Header>literalHeader: literalValue</Header><Header>storedHeader: StoredHeaderDefault</Header></Headers><Content></Content></Row>
 </Dataset>
 <Dataset name='proxyResult'>
- <Row><Method>GET</Method><UrlPath>/WsSmc/HttpEcho</UrlPath><UrlParameters>name=doe,joe&amp;number=1</UrlParameters><Headers><Header>Accept-Encoding: gzip, deflate</Header><Header>Accept: text/xml</Header><Header>constHeader: constHeaderValue</Header><Header>literalHeader: literalValue</Header><Header>storedHeader: StoredHeaderDefault</Header></Headers><Content></Content></Row>
+ <Row><Method>GET</Method><UrlPath>/WsSmc/HttpEcho</UrlPath><UrlParameters>name=doe,joe&amp;number=1</UrlParameters><Headers><Header>Accept-Encoding: gzip, deflate</Header><Header>Accept: text/xml</Header><Header>HPCC-Caller-Id: http222</Header><Header>HPCC-Global-Id: 9876543210</Header><Header>constHeader: constHeaderValue</Header><Header>literalHeader: literalValue</Header><Header>storedHeader: StoredHeaderDefault</Header></Headers><Content></Content></Row>
 </Dataset>

testing/regress/ecl/soapcall.ecl

@@ -39,13 +39,13 @@ ServiceOutRecord :=
     END;
 
 // simple query->dataset form
-output(SORT(SOAPCALL(targetURL,'soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord), LOG('simple')),record));
+output(SORT(SOAPCALL(targetURL,'soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord), LOG('simple'), HTTPHEADER('HPCC-Global-Id','12345678900'), HTTPHEADER('HPCC-Caller-Id','1111')),record));
 
 // double query->dataset form
-output(SORT(SOAPCALL(doubleTargetURL,'soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord)),record));
+output(SORT(SOAPCALL(doubleTargetURL,'soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord), HTTPHEADER('HPCC-Global-Id','12345678900'), HTTPHEADER('HPCC-Caller-Id','2222')),record));
 
 // simple dataset->dataset form
-output(sort(SOAPCALL(d, targetURL,'soapbase', { unkname }, DATASET(ServiceOutRecord)),record));
+output(sort(SOAPCALL(d, targetURL,'soapbase', { unkname }, DATASET(ServiceOutRecord), HTTPHEADER('HPCC-Global-Id','12345678900'), HTTPHEADER('HPCC-Caller-Id','3333')),record));
 
 // double query->dataset form
 ServiceOutRecord doError(d l) := TRANSFORM
@@ -73,7 +73,7 @@ END;
 // Test some failure cases
 
 output(SORT(SOAPCALL(d, blacklistedTargetURL,'soapbase', { unkname }, DATASET(ServiceOutRecord), onFail(doError(LEFT)),RETRY(0), log('SOAP: ' + unkname),TIMEOUT(1)), record));
-output(SORT(SOAPCALL(targetURL,'soapbase', { string unkname := 'FAIL' }, dataset(ServiceOutRecord),onFail(doError2),RETRY(0), LOG(MIN)),record));
+output(SORT(SOAPCALL(targetURL,'soapbase', { string unkname := 'FAIL' }, dataset(ServiceOutRecord),onFail(doError2),RETRY(0), LOG(MIN), HTTPHEADER('HPCC-Global-Id','12345678900'), HTTPHEADER('HPCC-Caller-Id','4444')),record));
 output(SORT(SOAPCALL(d, targetURL,'soapbaseNOSUCHQUERY', { unkname }, DATASET(ServiceOutRecord), onFail(doError3(LEFT)),MERGE(25),PARALLEL(1),RETRY(0), LOG(MIN)), record));
 
 childRecord := record

thorlcr/activities/soapcall/thsoapcallslave.cpp

@@ -61,10 +61,10 @@ public:
             switch (container.getKind())
             {
                 case TAKsoap_rowdataset:
-                    wscHelper.setown(createSoapCallHelper(this, queryRowAllocator(), authToken.str(), SCrow, NULL, queryDummyContextLogger(), NULL));
+                    wscHelper.setown(createSoapCallHelper(this, queryRowAllocator(), authToken.str(), SCrow, NULL, container.queryJob().queryContextLogger(), NULL));
                     break;
                 case TAKhttp_rowdataset:
-                    wscHelper.setown(createHttpCallHelper(this, queryRowAllocator(), authToken.str(), SCrow, NULL, queryDummyContextLogger(), NULL));
+                    wscHelper.setown(createHttpCallHelper(this, queryRowAllocator(), authToken.str(), SCrow, NULL, container.queryJob().queryContextLogger(), NULL));
                     break;
                 default:
                     throwUnexpected();
@@ -153,7 +153,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
         eof = false;
-        wscHelper.setown(createSoapCallHelper(this, queryRowAllocator(), authToken.str(), SCdataset, NULL, queryDummyContextLogger(),NULL));
+        wscHelper.setown(createSoapCallHelper(this, queryRowAllocator(), authToken.str(), SCdataset, NULL, container.queryJob().queryContextLogger(), NULL));
         wscHelper->start();
     }
     virtual void stop() override
@@ -242,7 +242,7 @@ public:
     {
         if (container.queryLocalOrGrouped() || firstNode())
         {
-            wscHelper.setown(createSoapCallHelper(this, NULL, authToken.str(), SCrow, NULL, queryDummyContextLogger(),NULL));
+            wscHelper.setown(createSoapCallHelper(this, NULL, authToken.str(), SCrow, NULL, container.queryJob().queryContextLogger(),NULL));
             wscHelper->start();
             wscHelper->waitUntilDone();
             IException *e = wscHelper->getError();
@@ -300,7 +300,7 @@ public:
 
         processed = THORDATALINK_STARTED;
 
-        wscHelper.setown(createSoapCallHelper(this, NULL, authToken.str(), SCdataset, NULL, queryDummyContextLogger(),NULL));
+        wscHelper.setown(createSoapCallHelper(this, NULL, authToken.str(), SCdataset, NULL, container.queryJob().queryContextLogger(),NULL));
         wscHelper->start();
         wscHelper->waitUntilDone();
         IException *e = wscHelper->getError();

thorlcr/graph/thgraph.cpp

@@ -2529,12 +2529,18 @@ class CThorContextLogger : implements IContextLogger, public CSimpleInterface
 {
     CJobBase &job;
     unsigned traceLevel;
+    StringAttr globalIdHeader;
+    StringAttr callerIdHeader;
+    StringAttr globalId;
+    StringBuffer localId;
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
     CThorContextLogger(CJobBase &_job) : job(_job)
     {
         traceLevel = 1;
+        if (globals->hasProp("@httpGlobalIdHeader"))
+            setHttpIdHeaders(globals->queryProp("@httpGlobalIdHeader"), globals->queryProp("@httpCallerIdHeader"));
     }
     virtual void CTXLOGva(const char *format, va_list args) const __attribute__((format(printf,2,0)))
     {
@@ -2566,6 +2572,35 @@ public:
     {
         return traceLevel;
     }
+    virtual void setGlobalId(const char *id, SocketEndpoint &ep, unsigned pid)
+    {
+        globalId.set(id);
+        appendLocalId(localId.clear(), ep, pid);
+    }
+    virtual const char *queryGlobalId() const
+    {
+        return globalId.get();
+    }
+    virtual const char *queryLocalId() const
+    {
+        return localId.str();
+    }
+    virtual void setHttpIdHeaders(const char *global, const char *caller)
+    {
+        if (global && *global)
+            globalIdHeader.set(global);
+        if (caller && *caller)
+            callerIdHeader.set(caller);
+    }
+    virtual const char *queryGlobalIdHttpHeader() const
+    {
+        return globalIdHeader.str();
+    }
+    virtual const char *queryCallerIdHttpHeader() const
+    {
+        return callerIdHeader.str();
+    }
+
 };
 
 ////

thorlcr/graph/thgraphmaster.cpp

@@ -1297,6 +1297,26 @@ CJobMaster::CJobMaster(IConstWorkUnit &_workunit, const char *graphName, ILoaded
     numChannels = 1;
     init();
 
+    if (workunit->hasDebugValue("GlobalId"))
+    {
+        SCMStringBuffer txId;
+        workunit->getDebugValue("GlobalId", txId);
+        if (txId.length())
+        {
+            SocketEndpoint thorEp;
+            thorEp.setLocalHost(getMachinePortBase());
+            logctx->setGlobalId(txId.str(), thorEp, 0);
+
+            VStringBuffer msg("GlobalId: %s", txId.str());
+            workunit->getDebugValue("CallerId", txId);
+            if (txId.length())
+                msg.append(", CallerId: ").append(txId.str());
+            txId.set(logctx->queryLocalId());
+            if (txId.length())
+                msg.append(", LocalId: ").append(txId.str());
+            logctx->CTXLOG("%s", msg.str());
+        }
+    }
     resumed = WUActionResume == workunit->getAction();
     fatalHandler.setown(new CFatalHandler(globals->getPropInt("@fatal_timeout", FATAL_TIMEOUT)));
     querySent = spillsSaved = false;

thorlcr/graph/thgraphslave.cpp

@@ -1491,6 +1491,26 @@ CJobSlave::CJobSlave(ISlaveWatchdog *_watchdog, IPropertyTree *_workUnitInfo, co
 
     init();
 
+    if (workUnitInfo->hasProp("Debug/globalid"))
+    {
+        const char *globalId = workUnitInfo->queryProp("Debug/globalid");
+        if (globalId && *globalId)
+        {
+            SocketEndpoint thorEp;
+            thorEp.setLocalHost(getMachinePortBase());
+            logctx->setGlobalId(globalId, thorEp, 0);
+
+            VStringBuffer msg("GlobalId: %s", globalId);
+            const char *callerId = workUnitInfo->queryProp("debug/callerid");
+            if (callerId && *callerId)
+                msg.append(", CallerId: ").append(callerId);
+            const char *localId = logctx->queryLocalId();
+            if (localId && *localId)
+                msg.append(", LocalId: ").append(localId);
+            logctx->CTXLOG("%s", msg.str());
+        }
+    }
+
     oldNodeCacheMem = 0;
     mpJobTag = _mpJobTag;
     slavemptag = _slavemptag;