Merge pull request #10705 from afishbeck/globalId

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

Reviewed-By: Richard Chapman <rchapman@hpccsystems.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();
@@ -1149,6 +1175,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;