-
Notifications
You must be signed in to change notification settings - Fork 2.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feature: add skywalking plugin. (#1241)
- Loading branch information
Showing
18 changed files
with
698 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| -- | ||
| -- Licensed to the Apache Software Foundation (ASF) under one or more | ||
| -- contributor license agreements. See the NOTICE file distributed with | ||
| -- this work for additional information regarding copyright ownership. | ||
| -- The ASF licenses this file to You 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. | ||
| -- | ||
| local core = require("apisix.core") | ||
| local ngx = ngx | ||
| local math = math | ||
|
|
||
| local sw_client = require("apisix.plugins.skywalking.client") | ||
| local sw_tracer = require("apisix.plugins.skywalking.tracer") | ||
|
|
||
| local plugin_name = "skywalking" | ||
|
|
||
|
|
||
| local schema = { | ||
| type = "object", | ||
| properties = { | ||
| endpoint = {type = "string"}, | ||
| sample_ratio = {type = "number", minimum = 0.00001, maximum = 1, default = 1} | ||
| }, | ||
| service_name = { | ||
| type = "string", | ||
| description = "service name for skywalking", | ||
| default = "APISIX", | ||
| }, | ||
| required = {"endpoint"} | ||
| } | ||
|
|
||
|
|
||
| local _M = { | ||
| version = 0.1, | ||
| priority = -1100, -- last running plugin, but before serverless post func | ||
| name = plugin_name, | ||
| schema = schema, | ||
| } | ||
|
|
||
|
|
||
| function _M.check_schema(conf) | ||
| return core.schema.check(schema, conf) | ||
| end | ||
|
|
||
|
|
||
| function _M.rewrite(conf, ctx) | ||
| core.log.debug("rewrite phase of skywalking plugin") | ||
| ctx.skywalking_sample = false | ||
| if conf.sample_ratio == 1 or math.random() < conf.sample_ratio then | ||
| ctx.skywalking_sample = true | ||
| sw_client.heartbeat(conf) | ||
| -- Currently, we can not have the upstream real network address | ||
| sw_tracer.start(ctx, conf.endpoint, "upstream service") | ||
| end | ||
| end | ||
|
|
||
|
|
||
| function _M.body_filter(conf, ctx) | ||
| if ctx.skywalking_sample and ngx.arg[2] then | ||
| sw_tracer.finish(ctx) | ||
| end | ||
| end | ||
|
|
||
|
|
||
| function _M.log(conf, ctx) | ||
| if ctx.skywalking_sample then | ||
| sw_tracer.prepareForReport(ctx, conf.endpoint) | ||
| end | ||
| end | ||
|
|
||
| return _M |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,226 @@ | ||
| -- | ||
| -- Licensed to the Apache Software Foundation (ASF) under one or more | ||
| -- contributor license agreements. See the NOTICE file distributed with | ||
| -- this work for additional information regarding copyright ownership. | ||
| -- The ASF licenses this file to You 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. | ||
| -- | ||
| local core = require("apisix.core") | ||
| local http = require("resty.http") | ||
| local cjson = require('cjson') | ||
| local ngx = ngx | ||
| local ipairs = ipairs | ||
|
|
||
| local register = require("skywalking.register") | ||
|
|
||
| local _M = {} | ||
|
|
||
| local function register_service(conf) | ||
| local endpoint = conf.endpoint | ||
|
|
||
| local tracing_buffer = ngx.shared['skywalking-tracing-buffer'] | ||
| local service_id = tracing_buffer:get(endpoint .. '_service_id') | ||
| if service_id then | ||
| return service_id | ||
| end | ||
|
|
||
| local service_name = conf.service_name | ||
| local service = register.newServiceRegister(service_name) | ||
|
|
||
| local httpc = http.new() | ||
| local res, err = httpc:request_uri(endpoint .. '/v2/service/register', | ||
| { | ||
| method = "POST", | ||
| body = core.json.encode(service), | ||
| headers = { | ||
| ["Content-Type"] = "application/json", | ||
| }, | ||
| }) | ||
| if not res then | ||
| core.log.error("skywalking service register failed, request uri: ", | ||
| endpoint .. '/v2/service/register', ", err: ", err) | ||
|
|
||
| elseif res.status == 200 then | ||
| core.log.debug("skywalking service register response: ", res.body) | ||
| local register_results = cjson.decode(res.body) | ||
|
|
||
| for _, result in ipairs(register_results) do | ||
| if result.key == service_name then | ||
| service_id = result.value | ||
| core.log.debug("skywalking service registered, service id:" | ||
| .. service_id) | ||
| end | ||
| end | ||
|
|
||
| else | ||
| core.log.error("skywalking service register failed, request uri:", | ||
| endpoint .. "/v2/service/register", | ||
| ", response code:", res.status) | ||
| end | ||
|
|
||
| if service_id then | ||
| tracing_buffer:set(endpoint .. '_service_id', service_id) | ||
| end | ||
|
|
||
| return service_id | ||
| end | ||
|
|
||
| local function register_service_instance(conf, service_id) | ||
| local endpoint = conf.endpoint | ||
|
|
||
| local tracing_buffer = ngx.shared['skywalking-tracing-buffer'] | ||
| local instance_id = tracing_buffer:get(endpoint .. '_instance_id') | ||
| if instance_id then | ||
| return instance_id | ||
| end | ||
|
|
||
| local service_instance_name = core.id.get() | ||
| local service_instance = register.newServiceInstanceRegister( | ||
| service_id, | ||
| service_instance_name, | ||
| ngx.now() * 1000) | ||
|
|
||
| local httpc = http.new() | ||
| local res, err = httpc:request_uri(endpoint .. '/v2/instance/register', | ||
| { | ||
| method = "POST", | ||
| body = core.json.encode(service_instance), | ||
| headers = { | ||
| ["Content-Type"] = "application/json", | ||
| }, | ||
| }) | ||
|
|
||
| if not res then | ||
| core.log.error("skywalking service Instance register failed", | ||
| ", request uri: ", conf.endpoint .. '/v2/instance/register', | ||
| ", err: ", err) | ||
|
|
||
| elseif res.status == 200 then | ||
| core.log.debug("skywalking service instance register response: ", res.body) | ||
| local register_results = cjson.decode(res.body) | ||
|
|
||
| for _, result in ipairs(register_results) do | ||
| if result.key == service_instance_name then | ||
| instance_id = result.value | ||
| core.log.debug("skywalking service Instance registered, ", | ||
| "service instance id: ", instance_id) | ||
| end | ||
| end | ||
|
|
||
| else | ||
| core.log.error("skywalking service instance register failed, ", | ||
| "response code:", res.status) | ||
| end | ||
|
|
||
| if instance_id then | ||
| tracing_buffer:set(endpoint .. '_instance_id', instance_id) | ||
| end | ||
|
|
||
| return instance_id | ||
| end | ||
|
|
||
| local function ping(endpoint) | ||
| local tracing_buffer = ngx.shared['skywalking-tracing-buffer'] | ||
| local ping_pkg = register.newServiceInstancePingPkg( | ||
| tracing_buffer:get(endpoint .. '_instance_id'), | ||
| core.id.get(), | ||
| ngx.now() * 1000) | ||
|
|
||
| local httpc = http.new() | ||
| local _, err = httpc:request_uri(endpoint .. '/v2/instance/heartbeat', { | ||
| method = "POST", | ||
| body = core.json.encode(ping_pkg), | ||
| headers = { | ||
| ["Content-Type"] = "application/json", | ||
| }, | ||
| }) | ||
|
|
||
| if err then | ||
| core.log.error("skywalking agent ping failed, err: ", err) | ||
| end | ||
| end | ||
|
|
||
| -- report trace segments to the backend | ||
| local function report_traces(endpoint) | ||
| local tracing_buffer = ngx.shared['skywalking-tracing-buffer'] | ||
| local segment = tracing_buffer:rpop(endpoint .. '_segment') | ||
|
|
||
| local count = 0 | ||
|
|
||
| local httpc = http.new() | ||
|
|
||
| while segment ~= nil do | ||
| local res, err = httpc:request_uri(endpoint .. '/v2/segments', { | ||
| method = "POST", | ||
| body = segment, | ||
| headers = { | ||
| ["Content-Type"] = "application/json", | ||
| }, | ||
| }) | ||
|
|
||
| if err == nil then | ||
| if res.status ~= 200 then | ||
| core.log.error("skywalking segment report failed, response code ", res.status) | ||
| break | ||
| else | ||
| count = count + 1 | ||
| end | ||
| else | ||
| core.log.error("skywalking segment report failed, err: ", err) | ||
| break | ||
| end | ||
|
|
||
| segment = tracing_buffer:rpop('segment') | ||
| end | ||
|
|
||
| if count > 0 then | ||
| core.log.debug(count, " skywalking segments reported") | ||
| end | ||
| end | ||
|
|
||
| do | ||
| local heartbeat_timer | ||
|
|
||
| function _M.heartbeat(conf) | ||
| local sw_heartbeat = function() | ||
| local service_id = register_service(conf) | ||
| if not service_id then | ||
| return | ||
| end | ||
|
|
||
| local service_instance_id = register_service_instance(conf, service_id) | ||
| if not service_instance_id then | ||
| return | ||
| end | ||
|
|
||
| report_traces(conf.endpoint) | ||
| ping(conf.endpoint) | ||
| end | ||
|
|
||
| local err | ||
| if ngx.worker.id() == 0 and not heartbeat_timer then | ||
| heartbeat_timer, err = core.timer.new("skywalking_heartbeat", | ||
| sw_heartbeat, | ||
| {check_interval = 3} | ||
| ) | ||
| if not heartbeat_timer then | ||
| core.log.error("failed to create skywalking_heartbeat timer: ", err) | ||
| else | ||
| core.log.info("succeed to create timer: skywalking heartbeat") | ||
| end | ||
| end | ||
| end | ||
|
|
||
| end -- do | ||
|
|
||
|
|
||
| return _M |
Oops, something went wrong.