Merge pull request #7 from mlixytz/master

Add support for 0.1.0

.gitignore

@@ -0,0 +1 @@
+.DS_Store

README.md

@@ -1,2 +1,3 @@
 # kong-plugin-skywalking
-The Nginx Lua agent 0.2.0 for Apache SkyWalking 8 kong-plugin
+
+The Nginx Lua agent 0.1.0 for Apache SkyWalking 7 kong-plugin

kong-plugin-skywalking-0.2.0-1.rockspec → kong-plugin-skywalking-0.1.0-1.rockspec

@@ -1,13 +1,13 @@
 package = "kong-plugin-skywalking" 
 
-version = "0.2.0-1" 
+version = "0.1.0-1" 
 
 local pluginName = "skywalking"
 
 supported_platforms = {"linux", "macosx"}
 source = {
   url = "https://github.com/polaris-liu/kong-plugin-skywalking.git",
-  tag = "0.2.0"
+  tag = "0.1.0"
 }
 
 description = {
@@ -26,7 +26,6 @@ build = {
     ["kong.plugins."..pluginName..".handler"] = "kong/plugins/"..pluginName.."/handler.lua",
     ["kong.plugins."..pluginName..".schema"] = "kong/plugins/"..pluginName.."/schema.lua",
     ["kong.plugins."..pluginName..".client"] = "kong/plugins/"..pluginName.."/client.lua",
-    ["kong.plugins."..pluginName..".correlation_context"] = "kong/plugins/"..pluginName.."/correlation_context.lua",
     ["kong.plugins."..pluginName..".management"] = "kong/plugins/"..pluginName.."/management.lua",
     ["kong.plugins."..pluginName..".segment"] = "kong/plugins/"..pluginName.."/segment.lua",
     ["kong.plugins."..pluginName..".segment_ref"] = "kong/plugins/"..pluginName.."/segment_ref.lua",

kong/plugins/skywalking/client.lua

@@ -15,6 +15,8 @@
 -- limitations under the License.
 --
 
+local timestamp = require "kong.tools.timestamp"
+
 local SEGMENT_BATCH_COUNT = 100
 
 local Client = {}
@@ -37,14 +39,25 @@ function Client:startBackendTimer(config)
 
     check = function(premature)
         if not premature then
-            local instancePropertiesSubmitted = metadata_buffer:get('sw_instancePropertiesSubmitted')
-            if (instancePropertiesSubmitted == nil or instancePropertiesSubmitted == false) then
-                self:reportServiceInstance(metadata_buffer, config)
-            else
-                self:ping(metadata_buffer, config)
+            local serviceId = metadata_buffer:get('serviceId')
+            if (serviceId == nil or serviceId == 0) then
+                self:registerService(metadata_buffer, config)
+            end
+
+            serviceId = metadata_buffer:get('serviceId')
+            if (serviceId ~= nil and serviceId ~= 0) then
+                local serviceInstId = metadata_buffer:get('serviceInstId')
+                if (serviceInstId == nil or serviceInstId == 0)  then
+                    self:registerServiceInstance(metadata_buffer, config)
+                end
             end
 
-            self:reportTraces(metadata_buffer, config)
+            -- After all register successfully, begin to send trace segments
+            local serviceInstId = metadata_buffer:get('serviceInstId')
+            if (serviceInstId ~= nil and serviceInstId ~= 0) then
+                self:reportTraces(metadata_buffer, config)
+                self:ping(metadata_buffer, config)
+            end
 
             -- do the health check
             local ok, err = new_timer(delay, check)
@@ -66,13 +79,55 @@ function Client:startBackendTimer(config)
     end
 end
 
-function Client:reportServiceInstance(metadata_buffer, config)
+-- Register service
+function Client:registerService(metadata_buffer, config)
 
-    local service_name = config.service_name
-    local service_instance_name = config.service_instance_name
+    local serviceName = config.service_name
+
+    local cjson = require('cjson')
+    local serviceRegister = require("kong.plugins.skywalking.management").newServiceRegister(serviceName)
+    local serviceRegisterParam = cjson.encode(serviceRegister)
+
+    local http = require('resty.http')
+    local httpc = http.new()
+    local res, err = httpc:request_uri(config.backend_http_uri .. '/v2/service/register', {
+        method = "POST",
+        body = serviceRegisterParam,
+        headers = {
+            ["Content-Type"] = "application/json",
+        },
+    })
+
+    if not res then
+        log(ERR, "Service register fails, " .. err)
+    elseif res.status == 200 then
+        log(DEBUG, "Service register response = " .. res.body)
+        local registerResults = cjson.decode(res.body)
+
+        for i, result in ipairs(registerResults)
+        do
+            if result.key == serviceName then
+                local serviceId = result.value
+                log(DEBUG, "Service registered, service id = " .. serviceId)
+                metadata_buffer:set('serviceId', serviceId)
+            end
+        end
+    else
+        log(ERR, "Service register fails, response code " .. res.status)
+    end
+end
+
+function Client:registerServiceInstance(metadata_buffer, config)
+
+    local service_id = metadata_buffer:get("serviceId")
+    local serviceInstName = 'NAME:' .. config.service_instance_name
+    metadata_buffer:set('serviceInstanceUUID', serviceInstName)
 
     local cjson = require('cjson')
-    local reportInstance = require("kong.plugins.skywalking.management").newReportInstanceProperties(service_name, service_instance_name)
+    local reportInstance = require("kong.plugins.skywalking.management").newServiceInstanceRegister(
+        service_id, 
+        serviceInstName,
+        timestamp.get_utc())
     local reportInstanceParam, err = cjson.encode(reportInstance)
     if err then
         log.err("Request to report instance fails, ", err)
@@ -81,7 +136,7 @@ function Client:reportServiceInstance(metadata_buffer, config)
 
     local http = require('resty.http')
     local httpc = http.new()
-    local uri = config.backend_http_uri .. '/v3/management/reportProperties'
+    local uri = config.backend_http_uri .. '/v2/instance/register'
 
     local res, err = httpc:request_uri(uri, {
         method = "POST",
@@ -92,31 +147,30 @@ function Client:reportServiceInstance(metadata_buffer, config)
     })
 
     if not res then
-        log.err("Instance report fails, uri:", uri, ", err:", err)
+        log.err("Service Instance register fails, uri:", uri, ", err:", err)
     elseif res.status == 200 then
-        log.debug("Instance report, uri:", uri, ", response = ", res.body)
-        metadata_buffer:set('sw_instancePropertiesSubmitted', true)
+        log.debug("Service Instance report, uri:", uri, ", response = ", res.body)
+        metadata_buffer:set('serviceInstId', serviceId)
     else
         log.err("Instance report fails, uri:", uri, ", response code ", res.status)
     end
 end
 
 -- Ping the backend to update instance heartheat
 function Client:ping(metadata_buffer, config)
-
-    local service_name = config.service_name
-    local service_instance_name = config.service_instance_name
-
     local cjson = require('cjson')
-    local pingPkg = require("kong.plugins.skywalking.management").newServiceInstancePingPkg(service_name, service_instance_name)
+    local pingPkg = require("kong.plugins.skywalking.management").newServiceInstancePingPkg(
+        metadata_buffer:get('serviceInstId'),
+        metadata_buffer:get('serviceInstanceUUID'),
+        timestamp.get_utc())
     local pingPkgParam, err = cjson.encode(pingPkg)
     if err then
         log.err("Agent ping fails, ", err)
     end
 
     local http = require('resty.http')
     local httpc = http.new()
-    local uri = config.backend_http_uri .. '/v3/management/keepAlive'
+    local uri = config.backend_http_uri .. '/v2/instance/heartbeat'
 
     local res, err = httpc:request_uri(uri, {
         method = "POST",
@@ -141,7 +195,7 @@ local function sendSegments(segmentTransform, backend_http_uri)
     local http = require('resty.http')
     local httpc = http.new()
 
-    local uri = backend_http_uri .. '/v3/segments'
+    local uri = backend_http_uri .. '/v2/segments'
 
     local res, err = httpc:request_uri(uri, {
         method = "POST",

kong/plugins/skywalking/management.lua

@@ -18,23 +18,52 @@
 local _M = {}
 
 -- Return Services as service register parameter
-function _M.newReportInstanceProperties(serviceName, serviceInstance)
+function _M.newServiceRegister(unRegisterServiceName)
+    local serv = {
+        services = {}
+    }
+
+    local service = {
+        serviceName = unRegisterServiceName,
+        -- Field type is optional, default value is `normal`
+        type = 'normal'
+    }
+
+    serv.services[#serv.services + 1] = service
+
+    return serv
+end
+
+-- Return Services as service register parameter
+function _M.newServiceInstanceRegister(registeredServiceId, serviceInstUUID, registerTime)
+    local serviceInstances = {
+        instances = {}
+    }
+
     local allProperties = {
         key = "language",
         value = "lua"
     }
 
-    return {
-        service = serviceName,
-        serviceInstance = serviceInstance,
-        properties = {allProperties}
+    local serviceInstance = {
+        serviceId = registeredServiceId,
+        instanceUUID = serviceInstUUID,
+        time = registerTime,
+        properties = {}
     }
+
+    serviceInstance.properties[#serviceInstance.properties + 1] = {allProperties}
+
+    serviceInstances.instances[#serviceInstances.instances + 1] = serviceInstance
+
+    return serviceInstances
 end
 
-function _M.newServiceInstancePingPkg(serviceName, serviceInstance)
+function _M.newServiceInstancePingPkg(serviceInstanceId, serviceInstanceUUID, updateTime)
     return {
-        service = serviceName,
-        serviceInstance = serviceInstance,
+        serviceInstanceId = serviceInstanceId,
+        time = updateTime,
+        serviceInstanceUUID = serviceInstanceUUID,
     }
 end