You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2021/05/24 13:45:39 UTC
[skywalking-kubernetes-event-exporter] 01/01: Fix Kubernetes
registry, signals overhaul
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 <ke...@apache.org>
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)
}