This is an automated email from the ASF dual-hosted git repository. kezhenxu94 pushed a commit to branch rel/1.0 in repository https://gitbox.apache.org/repos/asf/skywalking-kubernetes-event-exporter.git
commit c1e822ffeb9bc5bd236ccfb3366dc30488c83797 Author: kezhenxu94 <[email protected]> AuthorDate: Mon May 24 21:45:24 2021 +0800 Fix Kubernetes registry, signals overhaul This pull request includes: - Replace channel with context to pass signals. - Fix: sometimes events are not able to get service / instance because registry gets updates after events are sent. - Add licenses for dependencies --- CHANGES.md | 6 + README.md | 2 +- cmd/start.go | 13 +- configs/config.go | 20 +- configs/config_test.go | 3 +- dist/LICENSE | 252 ++++++++++++++++++++++++ dist/NOTICE | 5 + dist/licenses/LICENSE-golang-lru.txt | 362 +++++++++++++++++++++++++++++++++++ dist/licenses/LICENSE-logrus.txt | 21 ++ dist/licenses/LICENSE-yaml.v3.txt | 49 +++++ go.mod | 1 + go.sum | 3 +- licenses/LICENSE-linguist | 22 +++ pkg/event/event.go | 4 - pkg/exporter/console.go | 100 +++++----- pkg/exporter/exporter.go | 8 +- pkg/exporter/render.go | 91 +++++++++ pkg/exporter/skywalking.go | 165 +++++++--------- pkg/exporter/util.go | 36 ---- pkg/k8s/event.go | 20 +- pkg/k8s/registry.go | 181 ++++++++++-------- pkg/pipe/pipe.go | 57 ++---- 22 files changed, 1085 insertions(+), 336 deletions(-) diff --git a/CHANGES.md b/CHANGES.md new file mode 100644 index 0000000..a189589 --- /dev/null +++ b/CHANGES.md @@ -0,0 +1,6 @@ +# Change Log + +## 1.0 + +- Add Apache SkyWalking exporter to export events into SkyWalking OAP. +- Add console exporter for debugging purpose. diff --git a/README.md b/README.md index fd90561..8f19b53 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ [](https://github.com/apache/skywalking-kubernetes-event-exporter/actions/workflows/build-and-test.yaml) -<img src="http://skywalking.apache.org/assets/logo.svg"; alt="Sky Walking logo" height="90px" align="right" /> +<img src="https://skywalking.apache.org/assets/logo.svg"; alt="Sky Walking logo" height="90px" align="right" /> SkyWalking Kubernetes Event Exporter is able to watch, filter, and send Kubernetes events into [Apache SkyWalking](https://github.com/apache/skywalking) backend, afterwards, SkyWalking associates the events diff --git a/cmd/start.go b/cmd/start.go index cf7e624..9c112b2 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -20,6 +20,7 @@ package main import ( + "context" "os" "os/signal" "syscall" @@ -40,7 +41,11 @@ var startCmd = &cobra.Command{ Use: "start", Short: "Start skywalking-kubernetes-event-exporter", RunE: func(cmd *cobra.Command, args []string) error { - watcher, err := k8s.WatchEvents(v1.NamespaceAll) + ctx, cancel := context.WithCancel(context.Background()) + + addShutDownHook(cancel) + + watcher, err := k8s.WatchEvents(ctx, v1.NamespaceAll) if err != nil { return err } @@ -49,13 +54,11 @@ var startCmd = &cobra.Command{ Watcher: watcher, } - if err = p.Init(); err != nil { + if err = p.Init(ctx); err != nil { return err } - addShutDownHook(p.Stop) - - return p.Start() + return p.Start(ctx) }, } diff --git a/configs/config.go b/configs/config.go index 7243776..092230d 100644 --- a/configs/config.go +++ b/configs/config.go @@ -20,16 +20,17 @@ package configs import ( - "regexp" - "strings" + "context" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/k8s" "gopkg.in/yaml.v3" + v1 "k8s.io/api/core/v1" - "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" + "regexp" + "strings" - evnt "github.com/apache/skywalking-kubernetes-event-exporter/pkg/event" + "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" + "github.com/apache/skywalking-kubernetes-event-exporter/pkg/k8s" ) type FilterConfig struct { @@ -69,10 +70,7 @@ func (filter *FilterConfig) Init() { // Filter the given event with this filter instance. // Return true if the event is filtered, return false otherwise. -func (filter *FilterConfig) Filter(event *v1.Event) bool { - if event == evnt.Stopper { - return false - } +func (filter *FilterConfig) Filter(ctx context.Context, event *v1.Event) bool { if filter.Reason != "" && !filter.reasonRegExp.MatchString(event.Reason) { return true } @@ -98,8 +96,8 @@ func (filter *FilterConfig) Filter(event *v1.Event) bool { return true } if filter.Service != "" { - context := k8s.Registry.GetContext(event) - if svcName := strings.TrimSpace(context.Service.Name); !filter.serviceRegExp.MatchString(svcName) { + c := <-k8s.Registry.GetContext(ctx, event) + if svcName := strings.TrimSpace(c.Service.Name); !filter.serviceRegExp.MatchString(svcName) { return true } } diff --git a/configs/config_test.go b/configs/config_test.go index 288edf3..ec34b79 100644 --- a/configs/config_test.go +++ b/configs/config_test.go @@ -20,6 +20,7 @@ package configs import ( + "context" "testing" v1 "k8s.io/api/core/v1" @@ -98,7 +99,7 @@ func TestFilterConfig_Filter(t *testing.T) { Exporters: tt.fields.Exporters, } filter.Init() - if got := filter.Filter(tt.args.event); got != tt.want { + if got := filter.Filter(context.Background(), tt.args.event); got != tt.want { t.Errorf("Filter() = %v, want %v", got, tt.want) } }) diff --git a/dist/LICENSE b/dist/LICENSE new file mode 100644 index 0000000..beab566 --- /dev/null +++ b/dist/LICENSE @@ -0,0 +1,252 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + 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. + + +======================================================================= +Apache SkyWalking Eyes Subcomponents: + +The Apache SkyWalking Eyes project contains subcomponents with separate copyright +notices and license terms. Your use of the source code for the these +subcomponents is subject to the terms and conditions of the following +licenses. + + +======================================================================== +Apache 2.0 licenses +======================================================================== + +The following components are provided under the Apache License. See project link for details. +The text of each license is the standard Apache 2.0 license. + + cobra v1.1.3: https://github.com/spf13/cobra Apache 2.0 + grpc-go v1.36.1: https://github.com/grpc/grpc-go Apache 2.0 + k8s.io/api v0.20.5: https://github.com/kubernetes/api Apache 2.0 + k8s.io/client-go v0.20.5: https://github.com/kubernetes/client-go Apache 2.0 + +======================================================================== +MIT licenses +======================================================================== + +The following components are provided under the MIT license. See project link for details. +The text of each license is also included at licenses/LICENSE-[project].txt. + + logrus 1.8.1: https://github.com/sirupsen/logrus MIT + yaml v3: https://github.com/go-yaml/yaml/blob/v3 MIT + +======================================================================== +BSD licenses +======================================================================== + +The following components are provided under a BSD license. See project link for details. +The text of each license is also included at licenses/LICENSE-[project].txt. + + go-github v33: https://github.com/google/go-github BSD-3-Clause + oauth2 v0.0.0: https://github.com/golang/oauth2 BSD-3-Clause + +======================================================================== +MPL licenses +======================================================================== + +The following components are provided under MPL license. See project link for details. +The text of each license is also included at licenses/LICENSE-[project].txt. + + golang-lru v0.5.4: github.com/hashicorp/golang-lru MPL-2.0 diff --git a/dist/NOTICE b/dist/NOTICE new file mode 100644 index 0000000..e646f51 --- /dev/null +++ b/dist/NOTICE @@ -0,0 +1,5 @@ +Apache SkyWalking +Copyright 2017-2020 The Apache Software Foundation + +This product includes software developed at +The Apache Software Foundation (http://www.apache.org/). diff --git a/dist/licenses/LICENSE-golang-lru.txt b/dist/licenses/LICENSE-golang-lru.txt new file mode 100644 index 0000000..be2cc4d --- /dev/null +++ b/dist/licenses/LICENSE-golang-lru.txt @@ -0,0 +1,362 @@ +Mozilla Public License, version 2.0 + +1. Definitions + +1.1. "Contributor" + + means each individual or legal entity that creates, contributes to the + creation of, or owns Covered Software. + +1.2. "Contributor Version" + + means the combination of the Contributions of others (if any) used by a + Contributor and that particular Contributor's Contribution. + +1.3. "Contribution" + + means Covered Software of a particular Contributor. + +1.4. "Covered Software" + + means Source Code Form to which the initial Contributor has attached the + notice in Exhibit A, the Executable Form of such Source Code Form, and + Modifications of such Source Code Form, in each case including portions + thereof. + +1.5. "Incompatible With Secondary Licenses" + means + + a. that the initial Contributor has attached the notice described in + Exhibit B to the Covered Software; or + + b. that the Covered Software was made available under the terms of + version 1.1 or earlier of the License, but not also under the terms of + a Secondary License. + +1.6. "Executable Form" + + means any form of the work other than Source Code Form. + +1.7. "Larger Work" + + means a work that combines Covered Software with other material, in a + separate file or files, that is not Covered Software. + +1.8. "License" + + means this document. + +1.9. "Licensable" + + means having the right to grant, to the maximum extent possible, whether + at the time of the initial grant or subsequently, any and all of the + rights conveyed by this License. + +1.10. "Modifications" + + means any of the following: + + a. any file in Source Code Form that results from an addition to, + deletion from, or modification of the contents of Covered Software; or + + b. any new file in Source Code Form that contains any Covered Software. + +1.11. "Patent Claims" of a Contributor + + means any patent claim(s), including without limitation, method, + process, and apparatus claims, in any patent Licensable by such + Contributor that would be infringed, but for the grant of the License, + by the making, using, selling, offering for sale, having made, import, + or transfer of either its Contributions or its Contributor Version. + +1.12. "Secondary License" + + means either the GNU General Public License, Version 2.0, the GNU Lesser + General Public License, Version 2.1, the GNU Affero General Public + License, Version 3.0, or any later versions of those licenses. + +1.13. "Source Code Form" + + means the form of the work preferred for making modifications. + +1.14. "You" (or "Your") + + means an individual or a legal entity exercising rights under this + License. For legal entities, "You" includes any entity that controls, is + controlled by, or is under common control with You. For purposes of this + definition, "control" means (a) the power, direct or indirect, to cause + the direction or management of such entity, whether by contract or + otherwise, or (b) ownership of more than fifty percent (50%) of the + outstanding shares or beneficial ownership of such entity. + + +2. License Grants and Conditions + +2.1. Grants + + Each Contributor hereby grants You a world-wide, royalty-free, + non-exclusive license: + + a. under intellectual property rights (other than patent or trademark) + Licensable by such Contributor to use, reproduce, make available, + modify, display, perform, distribute, and otherwise exploit its + Contributions, either on an unmodified basis, with Modifications, or + as part of a Larger Work; and + + b. under Patent Claims of such Contributor to make, use, sell, offer for + sale, have made, import, and otherwise transfer either its + Contributions or its Contributor Version. + +2.2. Effective Date + + The licenses granted in Section 2.1 with respect to any Contribution + become effective for each Contribution on the date the Contributor first + distributes such Contribution. + +2.3. Limitations on Grant Scope + + The licenses granted in this Section 2 are the only rights granted under + this License. No additional rights or licenses will be implied from the + distribution or licensing of Covered Software under this License. + Notwithstanding Section 2.1(b) above, no patent license is granted by a + Contributor: + + a. for any code that a Contributor has removed from Covered Software; or + + b. for infringements caused by: (i) Your and any other third party's + modifications of Covered Software, or (ii) the combination of its + Contributions with other software (except as part of its Contributor + Version); or + + c. under Patent Claims infringed by Covered Software in the absence of + its Contributions. + + This License does not grant any rights in the trademarks, service marks, + or logos of any Contributor (except as may be necessary to comply with + the notice requirements in Section 3.4). + +2.4. Subsequent Licenses + + No Contributor makes additional grants as a result of Your choice to + distribute the Covered Software under a subsequent version of this + License (see Section 10.2) or under the terms of a Secondary License (if + permitted under the terms of Section 3.3). + +2.5. Representation + + Each Contributor represents that the Contributor believes its + Contributions are its original creation(s) or it has sufficient rights to + grant the rights to its Contributions conveyed by this License. + +2.6. Fair Use + + This License is not intended to limit any rights You have under + applicable copyright doctrines of fair use, fair dealing, or other + equivalents. + +2.7. Conditions + + Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted in + Section 2.1. + + +3. Responsibilities + +3.1. Distribution of Source Form + + All distribution of Covered Software in Source Code Form, including any + Modifications that You create or to which You contribute, must be under + the terms of this License. You must inform recipients that the Source + Code Form of the Covered Software is governed by the terms of this + License, and how they can obtain a copy of this License. You may not + attempt to alter or restrict the recipients' rights in the Source Code + Form. + +3.2. Distribution of Executable Form + + If You distribute Covered Software in Executable Form then: + + a. such Covered Software must also be made available in Source Code Form, + as described in Section 3.1, and You must inform recipients of the + Executable Form how they can obtain a copy of such Source Code Form by + reasonable means in a timely manner, at a charge no more than the cost + of distribution to the recipient; and + + b. You may distribute such Executable Form under the terms of this + License, or sublicense it under different terms, provided that the + license for the Executable Form does not attempt to limit or alter the + recipients' rights in the Source Code Form under this License. + +3.3. Distribution of a Larger Work + + You may create and distribute a Larger Work under terms of Your choice, + provided that You also comply with the requirements of this License for + the Covered Software. If the Larger Work is a combination of Covered + Software with a work governed by one or more Secondary Licenses, and the + Covered Software is not Incompatible With Secondary Licenses, this + License permits You to additionally distribute such Covered Software + under the terms of such Secondary License(s), so that the recipient of + the Larger Work may, at their option, further distribute the Covered + Software under the terms of either this License or such Secondary + License(s). + +3.4. Notices + + You may not remove or alter the substance of any license notices + (including copyright notices, patent notices, disclaimers of warranty, or + limitations of liability) contained within the Source Code Form of the + Covered Software, except that You may alter any license notices to the + extent required to remedy known factual inaccuracies. + +3.5. Application of Additional Terms + + You may choose to offer, and to charge a fee for, warranty, support, + indemnity or liability obligations to one or more recipients of Covered + Software. However, You may do so only on Your own behalf, and not on + behalf of any Contributor. You must make it absolutely clear that any + such warranty, support, indemnity, or liability obligation is offered by + You alone, and You hereby agree to indemnify every Contributor for any + liability incurred by such Contributor as a result of warranty, support, + indemnity or liability terms You offer. You may include additional + disclaimers of warranty and limitations of liability specific to any + jurisdiction. + +4. Inability to Comply Due to Statute or Regulation + + If it is impossible for You to comply with any of the terms of this License + with respect to some or all of the Covered Software due to statute, + judicial order, or regulation then You must: (a) comply with the terms of + this License to the maximum extent possible; and (b) describe the + limitations and the code they affect. Such description must be placed in a + text file included with all distributions of the Covered Software under + this License. Except to the extent prohibited by statute or regulation, + such description must be sufficiently detailed for a recipient of ordinary + skill to be able to understand it. + +5. Termination + +5.1. The rights granted under this License will terminate automatically if You + fail to comply with any of its terms. However, if You become compliant, + then the rights granted under this License from a particular Contributor + are reinstated (a) provisionally, unless and until such Contributor + explicitly and finally terminates Your grants, and (b) on an ongoing + basis, if such Contributor fails to notify You of the non-compliance by + some reasonable means prior to 60 days after You have come back into + compliance. Moreover, Your grants from a particular Contributor are + reinstated on an ongoing basis if such Contributor notifies You of the + non-compliance by some reasonable means, this is the first time You have + received notice of non-compliance with this License from such + Contributor, and You become compliant prior to 30 days after Your receipt + of the notice. + +5.2. If You initiate litigation against any entity by asserting a patent + infringement claim (excluding declaratory judgment actions, + counter-claims, and cross-claims) alleging that a Contributor Version + directly or indirectly infringes any patent, then the rights granted to + You by any and all Contributors for the Covered Software under Section + 2.1 of this License shall terminate. + +5.3. In the event of termination under Sections 5.1 or 5.2 above, all end user + license agreements (excluding distributors and resellers) which have been + validly granted by You or Your distributors under this License prior to + termination shall survive termination. + +6. Disclaimer of Warranty + + Covered Software is provided under this License on an "as is" basis, + without warranty of any kind, either expressed, implied, or statutory, + including, without limitation, warranties that the Covered Software is free + of defects, merchantable, fit for a particular purpose or non-infringing. + The entire risk as to the quality and performance of the Covered Software + is with You. Should any Covered Software prove defective in any respect, + You (not any Contributor) assume the cost of any necessary servicing, + repair, or correction. This disclaimer of warranty constitutes an essential + part of this License. No use of any Covered Software is authorized under + this License except under this disclaimer. + +7. Limitation of Liability + + Under no circumstances and under no legal theory, whether tort (including + negligence), contract, or otherwise, shall any Contributor, or anyone who + distributes Covered Software as permitted above, be liable to You for any + direct, indirect, special, incidental, or consequential damages of any + character including, without limitation, damages for lost profits, loss of + goodwill, work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses, even if such party shall have been + informed of the possibility of such damages. This limitation of liability + shall not apply to liability for death or personal injury resulting from + such party's negligence to the extent applicable law prohibits such + limitation. Some jurisdictions do not allow the exclusion or limitation of + incidental or consequential damages, so this exclusion and limitation may + not apply to You. + +8. Litigation + + Any litigation relating to this License may be brought only in the courts + of a jurisdiction where the defendant maintains its principal place of + business and such litigation shall be governed by laws of that + jurisdiction, without reference to its conflict-of-law provisions. Nothing + in this Section shall prevent a party's ability to bring cross-claims or + counter-claims. + +9. Miscellaneous + + This License represents the complete agreement concerning the subject + matter hereof. If any provision of this License is held to be + unenforceable, such provision shall be reformed only to the extent + necessary to make it enforceable. Any law or regulation which provides that + the language of a contract shall be construed against the drafter shall not + be used to construe this License against a Contributor. + + +10. Versions of the License + +10.1. New Versions + + Mozilla Foundation is the license steward. Except as provided in Section + 10.3, no one other than the license steward has the right to modify or + publish new versions of this License. Each version will be given a + distinguishing version number. + +10.2. Effect of New Versions + + You may distribute the Covered Software under the terms of the version + of the License under which You originally received the Covered Software, + or under the terms of any subsequent version published by the license + steward. + +10.3. Modified Versions + + If you create software not governed by this License, and you want to + create a new license for such software, you may create and use a + modified version of this License if you rename the license and remove + any references to the name of the license steward (except to note that + such modified license differs from this License). + +10.4. Distributing Source Code Form that is Incompatible With Secondary + Licenses If You choose to distribute Source Code Form that is + Incompatible With Secondary Licenses under the terms of this version of + the License, the notice described in Exhibit B of this License must be + attached. + +Exhibit A - Source Code Form License Notice + + This Source Code Form is subject to the + terms of the Mozilla Public License, v. + 2.0. If a copy of the MPL was not + distributed with this file, You can + obtain one at + http://mozilla.org/MPL/2.0/. + +If it is not possible or desirable to put the notice in a particular file, +then You may include the notice in a location (such as a LICENSE file in a +relevant directory) where a recipient would be likely to look for such a +notice. + +You may add additional accurate notices of copyright ownership. + +Exhibit B - "Incompatible With Secondary Licenses" Notice + + This Source Code Form is "Incompatible + With Secondary Licenses", as defined by + the Mozilla Public License, v. 2.0. diff --git a/dist/licenses/LICENSE-logrus.txt b/dist/licenses/LICENSE-logrus.txt new file mode 100644 index 0000000..f090cb4 --- /dev/null +++ b/dist/licenses/LICENSE-logrus.txt @@ -0,0 +1,21 @@ +The MIT License (MIT) + +Copyright (c) 2014 Simon Eskildsen + +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. diff --git a/dist/licenses/LICENSE-yaml.v3.txt b/dist/licenses/LICENSE-yaml.v3.txt new file mode 100644 index 0000000..5f6af8e --- /dev/null +++ b/dist/licenses/LICENSE-yaml.v3.txt @@ -0,0 +1,49 @@ +This project is covered by two different licenses: MIT and Apache. + +#### MIT License #### + +The following files were ported to Go from C files of libyaml, and thus +are still covered by their original MIT license, with the additional +copyright staring in 2011 when the project was ported over: + + apic.go emitterc.go parserc.go readerc.go scannerc.go + writerc.go yamlh.go yamlprivateh.go + +Copyright (c) 2006-2010 Kirill Simonov +Copyright (c) 2006-2011 Kirill Simonov + +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. + +### Apache License ### + +All the remaining project files are covered by the Apache license: + +Copyright (c) 2011-2019 Canonical Ltd + +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. diff --git a/go.mod b/go.mod index 579b967..650a495 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/apache/skywalking-kubernetes-event-exporter go 1.16 require ( + github.com/hashicorp/golang-lru v0.5.4 github.com/sirupsen/logrus v1.8.1 github.com/spf13/cobra v1.1.3 google.golang.org/grpc v1.36.1 diff --git a/go.sum b/go.sum index fe544d9..fad5406 100644 --- a/go.sum +++ b/go.sum @@ -183,8 +183,9 @@ github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= -github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU= github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= +github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc= +github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64= github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ= diff --git a/licenses/LICENSE-linguist b/licenses/LICENSE-linguist new file mode 100644 index 0000000..acc8e6f --- /dev/null +++ b/licenses/LICENSE-linguist @@ -0,0 +1,22 @@ +Copyright (c) 2017 GitHub, Inc. + +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. diff --git a/pkg/event/event.go b/pkg/event/event.go index b461205..b2dd8d2 100644 --- a/pkg/event/event.go +++ b/pkg/event/event.go @@ -19,12 +19,8 @@ package event -import v1 "k8s.io/api/core/v1" - type Source struct { Service string `mapstructure:"service"` ServiceInstance string `mapstructure:"serviceInstance"` Endpoint string `mapstructure:"endpoint"` } - -var Stopper = (*v1.Event)(nil) diff --git a/pkg/exporter/console.go b/pkg/exporter/console.go index 05a5b1f..c32fa27 100644 --- a/pkg/exporter/console.go +++ b/pkg/exporter/console.go @@ -20,22 +20,26 @@ package exporter import ( + "context" "encoding/json" + "fmt" + + "time" + + "github.com/sirupsen/logrus" sw "skywalking.apache.org/repo/goapi/collect/event/v3" k8score "k8s.io/api/core/v1" "github.com/apache/skywalking-kubernetes-event-exporter/configs" "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/event" ) // Console Exporter exports the events into console logs, this exporter is typically // used for debugging. type Console struct { - config ConsoleConfig - stopper chan struct{} + config ConsoleConfig } type ConsoleConfig struct { @@ -43,13 +47,11 @@ type ConsoleConfig struct { } func init() { - s := &Console{ - stopper: make(chan struct{}), - } + s := &Console{} RegisterExporter(s.Name(), s) } -func (exporter *Console) Init() error { +func (exporter *Console) Init(context.Context) error { config := ConsoleConfig{} if c := configs.GlobalConfig.Exporters[exporter.Name()]; c == nil { @@ -73,49 +75,59 @@ func (exporter *Console) Name() string { return "console" } -func (exporter *Console) Export(events chan *k8score.Event) { +func (exporter *Console) Export(ctx context.Context, events chan *k8score.Event) { logger.Log.Debugf("exporting events into %+v", exporter.Name()) - func() { - for { - select { - case <-exporter.stopper: - drain(events) - return - case kEvent := <-events: - if kEvent == event.Stopper { - return + for { + select { + case <-ctx.Done(): + logger.Log.Debugf("stopping exporter %+v", exporter.Name()) + return + case kEvent := <-events: + if logger.Log.IsLevelEnabled(logrus.DebugLevel) { + if bytes, err := json.Marshal(kEvent); err == nil { + logger.Log.Debugf("exporting event to %v: %v", exporter.Name(), string(bytes)) } - logger.Log.Debugf("exporting event to %v: %v", exporter.Name(), kEvent) + } - t := sw.Type_Normal - if kEvent.Type == "Warning" { - t = sw.Type_Error - } - swEvent := &sw.Event{ - Uuid: string(kEvent.UID), - Source: &sw.Source{}, - Name: kEvent.Reason, - Type: t, - Message: kEvent.Message, - StartTime: kEvent.FirstTimestamp.UnixNano() / 1000000, - EndTime: kEvent.LastTimestamp.Unix() / 1000000, - } - if exporter.config.Template != nil { - exporter.config.Template.render(swEvent, kEvent) - logger.Log.Debugf("rendered event is: %+v", swEvent) - } - if bytes, err := json.Marshal(swEvent); err != nil { - logger.Log.Errorf("failed to send event to %+v, %+v", exporter.Name(), err) - } else { - logger.Log.Infoln(string(bytes)) - } + t := sw.Type_Normal + if kEvent.Type == k8score.EventTypeWarning { + t = sw.Type_Error + } + swEvent := &sw.Event{ + Uuid: string(kEvent.UID), + Source: &sw.Source{}, + Name: kEvent.Reason, + Type: t, + Message: kEvent.Message, + StartTime: kEvent.FirstTimestamp.UnixNano() / 1000000, + EndTime: kEvent.LastTimestamp.Unix() / 1000000, + } + if exporter.config.Template != nil { + go func() { + renderCtx, cancel := context.WithTimeout(ctx, time.Minute) + done := exporter.config.Template.render(renderCtx, swEvent, kEvent) + select { + case <-done: + logger.Log.Debugf("rendered event is: %+v", swEvent) + exporter.export(swEvent) + case <-renderCtx.Done(): + logger.Log.Debugf("rendered event is: %+v", swEvent) + exporter.export(swEvent) + } + cancel() + }() + } else { + exporter.export(swEvent) } } - }() + } } -func (exporter *Console) Stop() { - exporter.stopper <- struct{}{} - close(exporter.stopper) +func (exporter *Console) export(swEvent *sw.Event) { + if bytes, err := json.Marshal(swEvent); err != nil { + logger.Log.Errorf("failed to send event to %+v, %+v", exporter.Name(), err) + } else { + logger.Log.Infoln(string(bytes)) + } } diff --git a/pkg/exporter/exporter.go b/pkg/exporter/exporter.go index bc3ba15..a94f1c6 100644 --- a/pkg/exporter/exporter.go +++ b/pkg/exporter/exporter.go @@ -20,7 +20,8 @@ package exporter import ( - "html/template" + "context" + "text/template" v1 "k8s.io/api/core/v1" @@ -30,9 +31,8 @@ import ( type Exporter interface { Name() string - Init() error - Export(events chan *v1.Event) - Stop() + Init(ctx context.Context) error + Export(ctx context.Context, events chan *v1.Event) } var exporters = map[string]Exporter{} diff --git a/pkg/exporter/render.go b/pkg/exporter/render.go new file mode 100644 index 0000000..4df36f2 --- /dev/null +++ b/pkg/exporter/render.go @@ -0,0 +1,91 @@ +/* + * Licensed to 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. Apache Software Foundation (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. + */ + +package exporter + +import ( + "bytes" + "context" + "encoding/json" + "text/template" + + "github.com/sirupsen/logrus" + k8score "k8s.io/api/core/v1" + sw "skywalking.apache.org/repo/goapi/collect/event/v3" + + "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" + "github.com/apache/skywalking-kubernetes-event-exporter/pkg/k8s" +) + +func (tmplt *EventTemplate) render(ctx context.Context, swEvent *sw.Event, kEvent *k8score.Event) chan bool { + done := make(chan bool) + + go func() { + var templateCtx k8s.TemplateContext + + select { + case templateCtx = <-k8s.Registry.GetContext(ctx, kEvent): + case <-ctx.Done(): + done <- true + return + } + + if logger.Log.IsLevelEnabled(logrus.DebugLevel) { + if bs, err := json.Marshal(templateCtx); err == nil { + logger.Log.Debugf("template context %+v", string(bs)) + } + } + + render := func(t *template.Template, destination *string) error { + if t == nil { + return nil + } + + var buf bytes.Buffer + + if err := t.Execute(&buf, templateCtx); err != nil { + return err + } + + if buf.Len() > 0 { + *destination = buf.String() + } + + return nil + } + + if err := render(tmplt.messageTemplate, &swEvent.Message); err != nil { + logger.Log.Debugf("failed to render the template, using the default event content. %+v", err) + } + + if err := render(tmplt.sourceTemplate.serviceTemplate, &swEvent.Source.Service); err != nil { + logger.Log.Debugf("failed to render service template, using the default event content. %+v", err) + } + if err := render(tmplt.sourceTemplate.serviceInstanceTemplate, &swEvent.Source.ServiceInstance); err != nil { + logger.Log.Debugf("failed to render service instance template, using the default event content. %+v", err) + } + if err := render(tmplt.sourceTemplate.endpointTemplate, &swEvent.Source.Endpoint); err != nil { + logger.Log.Debugf("failed to render endpoin template, using the default event content. %+v", err) + } + + done <- true + }() + + return done +} diff --git a/pkg/exporter/skywalking.go b/pkg/exporter/skywalking.go index 7319bf4..aaf49fe 100644 --- a/pkg/exporter/skywalking.go +++ b/pkg/exporter/skywalking.go @@ -20,31 +20,26 @@ package exporter import ( - "bytes" "context" "encoding/json" "fmt" - "html/template" "time" - sw "skywalking.apache.org/repo/goapi/collect/event/v3" + "github.com/sirupsen/logrus" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/k8s" + sw "skywalking.apache.org/repo/goapi/collect/event/v3" "google.golang.org/grpc" k8score "k8s.io/api/core/v1" "github.com/apache/skywalking-kubernetes-event-exporter/configs" "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/event" ) // SkyWalking Exporter exports the events into Apache SkyWalking OAP server. type SkyWalking struct { - config SkyWalkingConfig - client sw.EventServiceClient - connection *grpc.ClientConn - stopper chan struct{} + config SkyWalkingConfig + client sw.EventServiceClient } type SkyWalkingConfig struct { @@ -53,13 +48,11 @@ type SkyWalkingConfig struct { } func init() { - s := &SkyWalking{ - stopper: make(chan struct{}), - } + s := &SkyWalking{} RegisterExporter(s.Name(), s) } -func (exporter *SkyWalking) Init() error { +func (exporter *SkyWalking) Init(ctx context.Context) error { config := SkyWalkingConfig{} if c := configs.GlobalConfig.Exporters[exporter.Name()]; c == nil { @@ -80,9 +73,16 @@ func (exporter *SkyWalking) Init() error { } exporter.config = config - exporter.connection = conn exporter.client = sw.NewEventServiceClient(conn) + go func() { + <-ctx.Done() + + if err := conn.Close(); err != nil { + logger.Log.Errorf("failed to close connection. %+v", err) + } + }() + return nil } @@ -90,109 +90,74 @@ func (exporter *SkyWalking) Name() string { return "skywalking" } -// TODO error handling -func (exporter *SkyWalking) Export(events chan *k8score.Event) { +func (exporter *SkyWalking) Export(ctx context.Context, events chan *k8score.Event) { logger.Log.Debugf("exporting events into %+v", exporter.Name()) - stream, err := exporter.client.Collect(context.Background()) + stream, err := exporter.client.Collect(ctx) + for err != nil { select { - case <-exporter.stopper: - drain(events) + case <-ctx.Done(): + logger.Log.Debugf("stopping exporter %+v", exporter.Name()) + if err = stream.CloseSend(); err != nil { + logger.Log.Warnf("failed to close stream. %+v", err) + } return default: logger.Log.Errorf("failed to connect to SkyWalking server. %+v", err) time.Sleep(3 * time.Second) - stream, err = exporter.client.Collect(context.Background()) + stream, err = exporter.client.Collect(ctx) } } - defer func() { - if err := stream.CloseSend(); err != nil { - logger.Log.Warnf("failed to close stream. %+v", err) - } - }() - - func() { - for { - select { - case <-exporter.stopper: - drain(events) - return - case kEvent := <-events: - if kEvent == event.Stopper { - return - } - logger.Log.Debugf("exporting event to %v: %v", exporter.Name(), kEvent) - - t := sw.Type_Normal - if kEvent.Type == "Warning" { - t = sw.Type_Error - } - swEvent := &sw.Event{ - Uuid: string(kEvent.UID), - Source: &sw.Source{}, - Name: kEvent.Reason, - Type: t, - Message: kEvent.Message, - StartTime: kEvent.FirstTimestamp.UnixNano() / 1000000, - EndTime: kEvent.LastTimestamp.Unix() / 1000000, - } - if exporter.config.Template != nil { - exporter.config.Template.render(swEvent, kEvent) - logger.Log.Debugf("rendered event is: %+v", swEvent) - } - if err := stream.Send(swEvent); err != nil { - logger.Log.Errorf("failed to send event to %+v. %+v", exporter.Name(), err) + for { + select { + case <-ctx.Done(): + logger.Log.Debugf("stopping exporter %+v", exporter.Name()) + return + case kEvent := <-events: + if logger.Log.IsLevelEnabled(logrus.DebugLevel) { + if bytes, err := json.Marshal(kEvent); err == nil { + logger.Log.Debugf("exporting event to %v: %v", exporter.Name(), string(bytes)) } } - } - }() -} - -func (tmplt *EventTemplate) render(swEvent *sw.Event, kEvent *k8score.Event) { - templateCtx := k8s.Registry.GetContext(kEvent) - - logger.Log.Debugf("template context %+v", templateCtx) - - render := func(t *template.Template, destination *string) error { - if t == nil { - return nil - } - - var buf bytes.Buffer - if err := t.Execute(&buf, templateCtx); err != nil { - return err - } - - if buf.Len() > 0 { - *destination = buf.String() + t := sw.Type_Normal + if kEvent.Type == k8score.EventTypeWarning { + t = sw.Type_Error + } + swEvent := &sw.Event{ + Uuid: string(kEvent.UID), + Source: &sw.Source{}, + Name: kEvent.Reason, + Type: t, + Message: kEvent.Message, + StartTime: kEvent.FirstTimestamp.UnixNano() / 1000000, + EndTime: kEvent.LastTimestamp.UnixNano() / 1000000, + } + if exporter.config.Template != nil { + go func() { + renderCtx, cancel := context.WithTimeout(ctx, time.Minute) + done := exporter.config.Template.render(renderCtx, swEvent, kEvent) + select { + case <-done: + logger.Log.Debugf("done: rendered event is: %+v", swEvent) + exporter.export(stream, swEvent) + case <-renderCtx.Done(): + logger.Log.Debugf("canceled: rendered event is: %+v", swEvent) + exporter.export(stream, swEvent) + } + cancel() + }() + } else { + exporter.export(stream, swEvent) + } } - - return nil - } - - if err := render(tmplt.messageTemplate, &swEvent.Message); err != nil { - logger.Log.Debugf("failed to render the template, using the default event content. %+v", err) - } - - if err := render(tmplt.sourceTemplate.serviceTemplate, &swEvent.Source.Service); err != nil { - logger.Log.Debugf("failed to render service template, using the default event content. %+v", err) - } - if err := render(tmplt.sourceTemplate.serviceInstanceTemplate, &swEvent.Source.ServiceInstance); err != nil { - logger.Log.Debugf("failed to render service instance template, using the default event content. %+v", err) - } - if err := render(tmplt.sourceTemplate.endpointTemplate, &swEvent.Source.Endpoint); err != nil { - logger.Log.Debugf("failed to render endpoin template, using the default event content. %+v", err) } } -func (exporter *SkyWalking) Stop() { - exporter.stopper <- struct{}{} - close(exporter.stopper) - - if err := exporter.connection.Close(); err != nil { - logger.Log.Errorf("failed to close connection. %+v", err) +func (exporter SkyWalking) export(stream sw.EventService_CollectClient, swEvent *sw.Event) { + if err := stream.Send(swEvent); err != nil { + logger.Log.Errorf("failed to send event to %+v. %+v", exporter.Name(), err) } } diff --git a/pkg/exporter/util.go b/pkg/exporter/util.go deleted file mode 100644 index b389f2d..0000000 --- a/pkg/exporter/util.go +++ /dev/null @@ -1,36 +0,0 @@ -/* - * Licensed to 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. Apache Software Foundation (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. - */ - -package exporter - -import ( - k8score "k8s.io/api/core/v1" - - "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/event" -) - -func drain(events chan *k8score.Event) { - logger.Log.Debugf("draining event channel") - for e := range events { - if e == event.Stopper { - break - } - } -} diff --git a/pkg/k8s/event.go b/pkg/k8s/event.go index 32a2b55..f8cbe8e 100644 --- a/pkg/k8s/event.go +++ b/pkg/k8s/event.go @@ -20,7 +20,10 @@ package k8s import ( + "context" + v1 "k8s.io/api/core/v1" + "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -31,7 +34,6 @@ import ( type EventWatcher struct { Events chan *v1.Event informer cache.SharedIndexInformer - stopCh chan struct{} } func (w EventWatcher) OnAdd(obj interface{}) { @@ -45,20 +47,19 @@ func (w EventWatcher) OnUpdate(_, newObj interface{}) { func (w EventWatcher) OnDelete(_ interface{}) { } -func (w EventWatcher) Start() { +func (w EventWatcher) Start(ctx context.Context) { logger.Log.Debugf("starting event watcher") - go w.informer.Run(w.stopCh) -} + go w.informer.Run(ctx.Done()) -func (w EventWatcher) Stop() { - logger.Log.Debugf("stopping event watcher") + go func() { + <-ctx.Done() - w.stopCh <- struct{}{} - close(w.stopCh) + logger.Log.Debugf("stopping event watcher") + }() } -func WatchEvents(ns string) (*EventWatcher, error) { +func WatchEvents(_ context.Context, ns string) (*EventWatcher, error) { config, err := GetConfig() if err != nil { return nil, err @@ -70,7 +71,6 @@ func WatchEvents(ns string) (*EventWatcher, error) { watcher := &EventWatcher{ informer: informer, Events: make(chan *v1.Event), - stopCh: make(chan struct{}), } informer.AddEventHandler(watcher) diff --git a/pkg/k8s/registry.go b/pkg/k8s/registry.go index cdf6760..3f2e831 100644 --- a/pkg/k8s/registry.go +++ b/pkg/k8s/registry.go @@ -20,14 +20,19 @@ package k8s import ( - "time" + "context" + "fmt" + + "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" + lru "github.com/hashicorp/golang-lru" corev1 "k8s.io/api/core/v1" + + "time" + "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" - - "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" ) type id struct { @@ -35,31 +40,35 @@ type id struct { name string } +func (i id) String() string { + return fmt.Sprintf("%+v:%+v", i.namespace, i.name) +} + type registry struct { informers []cache.SharedIndexInformer - stopCh chan struct{} - podIDIpMap map[id]string - idSvcMap map[id]*corev1.Service - idPodMap map[id]*corev1.Pod - ipSvcIDMap map[string]id + podIDIpMap *lru.Cache // map[id]string + idSvcMap *lru.Cache // map[id]*corev1.Service + idPodMap *lru.Cache // map[id]*corev1.Pod + ipSvcIDMap *lru.Cache // map[string]id } func (r registry) OnAdd(obj interface{}) { switch o := obj.(type) { case *corev1.Pod: - podID := id{namespace: o.Namespace, name: o.Name} - r.podIDIpMap[podID] = o.Status.PodIP - r.idPodMap[podID] = o + podID := id{namespace: o.Namespace, name: o.Name}.String() + + r.podIDIpMap.Add(podID, o.Status.PodIP) + r.idPodMap.Add(podID, o) case *corev1.Service: - r.idSvcMap[id{namespace: o.Namespace, name: o.Name}] = o + svcID := id{namespace: o.Namespace, name: o.Name}.String() + + r.idSvcMap.Add(svcID, o) case *corev1.Endpoints: for _, subset := range o.Subsets { for _, address := range subset.Addresses { - r.ipSvcIDMap[address.IP] = id{ - namespace: o.ObjectMeta.Namespace, - name: o.ObjectMeta.Name, - } + svcID := id{namespace: o.ObjectMeta.Namespace, name: o.ObjectMeta.Name}.String() + r.ipSvcIDMap.Add(address.IP, svcID) } } } @@ -70,98 +79,104 @@ func (r registry) OnUpdate(oldObj, newObj interface{}) { r.OnAdd(newObj) } -func (r registry) OnDelete(obj interface{}) { - switch o := obj.(type) { - case *corev1.Pod: - podID := id{namespace: o.Namespace, name: o.Name} - go func() { - time.Sleep(3 * time.Second) - delete(r.podIDIpMap, podID) - delete(r.idPodMap, podID) - }() - case *corev1.Service: - go func() { - time.Sleep(3 * time.Second) - delete(r.idSvcMap, id{namespace: o.Namespace, name: o.Name}) - }() - case *corev1.Endpoints: - go func() { - for _, subset := range o.Subsets { - for _, address := range subset.Addresses { - time.Sleep(3 * time.Second) - delete(r.ipSvcIDMap, address.IP) - } - } - }() - } +func (r registry) OnDelete(_ interface{}) { } -func (r *registry) Start() { +func (r *registry) Start(ctx context.Context) { logger.Log.Debugf("starting registry") for _, informer := range r.informers { - go informer.Run(r.stopCh) + go informer.Run(ctx.Done()) } } -func (r *registry) Stop() { - logger.Log.Debugf("stopping registry") - - r.stopCh <- struct{}{} - close(r.stopCh) -} - type TemplateContext struct { Service *corev1.Service Pod *corev1.Pod Event *corev1.Event } -func (r *registry) GetContext(e *corev1.Event) TemplateContext { - result := TemplateContext{Event: e} - - if obj := e.InvolvedObject; obj.Kind == "Pod" { - podID := id{ - namespace: obj.Namespace, - name: obj.Name, - } - podIP := r.podIDIpMap[podID] - svcID := r.ipSvcIDMap[podIP] - - result.Pod = r.idPodMap[podID] - result.Service = r.idSvcMap[svcID] - } +func (r *registry) GetContext(ctx context.Context, e *corev1.Event) chan TemplateContext { + resultCh := make(chan TemplateContext) - if obj := e.InvolvedObject; obj.Kind == "Service" { - svcID := id{ - namespace: obj.Namespace, - name: obj.Name, - } - result.Service = r.idSvcMap[svcID] + result := TemplateContext{ + Event: e, + Pod: &corev1.Pod{}, + Service: &corev1.Service{}, } - if result.Pod == nil { - result.Pod = &corev1.Pod{} - } - if result.Service == nil { - result.Service = &corev1.Service{} - } + go func() { + for { + select { + case <-ctx.Done(): + resultCh <- result + return + default: + if obj := e.InvolvedObject; obj.Kind == "Pod" { + podID := id{namespace: obj.Namespace, name: obj.Name}.String() + + pod, ok := r.idPodMap.Get(podID) + if !ok { + break + } + result.Pod = pod.(*corev1.Pod) + + podIP, ok := r.podIDIpMap.Get(podID) + if !ok { + break + } + + svcID, ok := r.ipSvcIDMap.Get(podIP.(string)) + if !ok { + break + } + + svc, ok := r.idSvcMap.Get(svcID.(string)) + if !ok { + break + } + result.Service = svc.(*corev1.Service) + } else if obj.Kind == "Service" { + svcID := id{namespace: obj.Namespace, name: obj.Name}.String() + + svc, ok := r.idSvcMap.Get(svcID) + if !ok { + break + } + result.Service = svc.(*corev1.Service) + } - return result -} + resultCh <- result -var Registry = ®istry{ - stopCh: make(chan struct{}), + return + } + time.Sleep(3 * time.Second) + } + }() - podIDIpMap: make(map[id]string), - idSvcMap: make(map[id]*corev1.Service), - idPodMap: make(map[id]*corev1.Pod), - ipSvcIDMap: make(map[string]id), + return resultCh } +var Registry = ®istry{} + func (r *registry) Init() error { logger.Log.Debugf("initializing template context registry") + var err error + + if Registry.podIDIpMap, err = lru.New(1000); err != nil { + return err + } + if Registry.idSvcMap, err = lru.New(1000); err != nil { + return err + } + if Registry.idPodMap, err = lru.New(1000); err != nil { + return err + } + if Registry.ipSvcIDMap, err = lru.New(1000); err != nil { + return err + } + config, err := GetConfig() if err != nil { return err diff --git a/pkg/pipe/pipe.go b/pkg/pipe/pipe.go index 5f32c00..e10aaf9 100644 --- a/pkg/pipe/pipe.go +++ b/pkg/pipe/pipe.go @@ -20,13 +20,14 @@ package pipe import ( + "context" "fmt" + "time" v1 "k8s.io/api/core/v1" "github.com/apache/skywalking-kubernetes-event-exporter/configs" "github.com/apache/skywalking-kubernetes-event-exporter/internal/pkg/logger" - "github.com/apache/skywalking-kubernetes-event-exporter/pkg/event" exp "github.com/apache/skywalking-kubernetes-event-exporter/pkg/exporter" "github.com/apache/skywalking-kubernetes-event-exporter/pkg/k8s" ) @@ -40,13 +41,11 @@ type workflow struct { type Pipe struct { Watcher *k8s.EventWatcher workflows []workflow - stopper chan struct{} } -func (p *Pipe) Init() error { +func (p *Pipe) Init(ctx context.Context) error { logger.Log.Debugf("initializing pipe") - p.stopper = make(chan struct{}) p.workflows = []workflow{} initialized := map[string]bool{} @@ -65,7 +64,7 @@ func (p *Pipe) Init() error { logger.Log.Debugf("exporter %+v has been initialized, skip", name) continue } - if err := exporter.Init(); err != nil { + if err := exporter.Init(ctx); err != nil { return err } initialized[name] = true @@ -89,45 +88,31 @@ func (p *Pipe) Init() error { return nil } -func (p *Pipe) Start() error { - p.Watcher.Start() +func (p *Pipe) Start(ctx context.Context) error { + p.Watcher.Start(ctx) - k8s.Registry.Start() + k8s.Registry.Start(ctx) for _, wkfl := range p.workflows { - go wkfl.exporter.Export(wkfl.events) + go wkfl.exporter.Export(ctx, wkfl.events) } - for stopped := false; !stopped; { + for { select { - case <-p.stopper: - for _, w := range p.workflows { - w.events <- event.Stopper - } - stopped = true + case <-ctx.Done(): + logger.Log.Debugf("stopping pipe") + return nil case e := <-p.Watcher.Events: - for _, w := range p.workflows { - if w.filter.Filter(e) { - continue - } - w.events <- e + for _, wkfl := range p.workflows { + go func(w workflow) { + fCtx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + + if !w.filter.Filter(fCtx, e) { + w.events <- e + } + }(wkfl) } } } - - return nil -} - -func (p *Pipe) Stop() { - p.Watcher.Stop() - - for _, w := range p.workflows { - logger.Log.Debugf("stopping exporter %+v.", w.exporter.Name()) - w.exporter.Stop() - } - - k8s.Registry.Stop() - - p.stopper <- struct{}{} - close(p.stopper) }
![http://skywalking.apache.org/assets/logo.svg"](http://skywalking.apache.org/assets/logo.svg"){kind=link}
![https://skywalking.apache.org/assets/logo.svg"](https://skywalking.apache.org/assets/logo.svg"){kind=link}