Skip to content

Commit

Permalink
ha buckets
Browse files Browse the repository at this point in the history
  • Loading branch information
lionelvillard committed Sep 14, 2020
1 parent 4873a50 commit adda82f
Show file tree
Hide file tree
Showing 17 changed files with 147 additions and 609 deletions.
2 changes: 1 addition & 1 deletion cmd/mtping/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ func main() {
// which under the cover delays the release of the lease.
ctx := mtping.NewDelayingContext(sctx, mtping.GetNoShutDownAfterValue())

ctx = adapter.WithInjectorEnabled(ctx)
ctx = adapter.WithController(ctx, mtping.NewController)
ctx = adapter.WithHAEnabled(ctx)
adapter.MainWithContext(ctx, component, mtping.NewEnvConfig, mtping.NewAdapter)
}
2 changes: 1 addition & 1 deletion config/core/deployments/pingsource-mt-adapter.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ metadata:
eventing.knative.dev/release: devel
spec:
# when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
replicas: 0
replicas: 3
selector:
matchLabels:
eventing.knative.dev/source: ping-source-controller
Expand Down
4 changes: 2 additions & 2 deletions docs/source/receive-adapter.md
Original file line number Diff line number Diff line change
Expand Up @@ -83,12 +83,12 @@ handle more than one source instance at a time, either all source instances in
one namespace or all source instances in the cluster.

In order to support multi-resources per receive adapter, you need to enable the
ConfigMap watcher feature, as follows:
Controller watcher feature, as follows:

```go
func main() {
ctx := signals.NewContext()
ctx = adapter.WithInjectorEnabled(ctx)
ctx = adapter.WithController(ctx, youradapter.NewController)
ctx = adapter.WithConfigMapWatcherEnabled(ctx)
adapter.MainWithContext(ctx, "yourcomponent",
youradapter.NewEnvConfig,
Expand Down
22 changes: 22 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -191,8 +191,12 @@ github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkY
github.com/asaskevich/govalidator v0.0.0-20180720115003-f9ffefc3facf/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
github.com/asaskevich/govalidator v0.0.0-20200108200545-475eaeb16496/go.mod h1:oGkLhpf+kjZl6xBf758TQhh5XrAeiJv/7FRz/2spLIg=
<<<<<<< HEAD
github.com/aws/aws-k8s-tester v0.0.0-20190114231546-b411acf57dfe/go.mod h1:1ADF5tAtU1/mVtfMcHAYSm2fPw71DA7fFk0yed64/0I=
github.com/aws/aws-k8s-tester v0.9.3/go.mod h1:nsh1f7joi8ZI1lvR+Ron6kJM2QdCYPU/vFePghSSuTc=
=======
github.com/aslakhellesoy/gox v1.0.100/go.mod h1:AJl542QsKKG96COVsv0N74HHzVQgDIQPceVUh1aeU2M=
>>>>>>> ha buckets
github.com/aws/aws-k8s-tester v1.0.0/go.mod h1:NUNd9k43+h9O5tvwL+4N1Ctb//SapmeeFX1G0/2/0Qc=
github.com/aws/aws-sdk-go v1.15.11/go.mod h1:mFuSZ37Z9YOHbQEwBWztmVzqXrEkub65tZoCYDt7FT0=
github.com/aws/aws-sdk-go v1.15.27/go.mod h1:mFuSZ37Z9YOHbQEwBWztmVzqXrEkub65tZoCYDt7FT0=
Expand Down Expand Up @@ -291,6 +295,13 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/cucumber/gherkin-go/v11 v11.0.0 h1:cwVwN1Qn2VRSfHZNLEh5x00tPBmZcjATBWDpxsR5Xug=
github.com/cucumber/gherkin-go/v11 v11.0.0/go.mod h1:CX33k2XU2qog4e+TFjOValoq6mIUq0DmVccZs238R9w=
github.com/cucumber/godog v0.10.0 h1:W01u1+o8bRpgqJRLrclN3iAanU1jAao+TwOMoSV9g1Y=
github.com/cucumber/godog v0.10.0/go.mod h1:0Q+MOUg8Z9AhzLV+nNMbThQ2x1b17yYwGyahApTLjJA=
github.com/cucumber/messages-go/v10 v10.0.1/go.mod h1:kA5T38CBlBbYLU12TIrJ4fk4wSkVVOgyh7Enyy8WnSg=
github.com/cucumber/messages-go/v10 v10.0.3 h1:m/9SD/K/A15WP7i1aemIv7cwvUw+viS51Ui5HBw1cdE=
github.com/cucumber/messages-go/v10 v10.0.3/go.mod h1:9jMZ2Y8ZxjLY6TG2+x344nt5rXstVVDYSdS5ySfI1WY=
github.com/cyphar/filepath-securejoin v0.2.2/go.mod h1:FpkQEhXnPnOthhzymB7CGsFk2G9VLXONKD9G7QGMM+4=
github.com/davecgh/go-spew v0.0.0-20151105211317-5215b55f46b2/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
Expand Down Expand Up @@ -471,6 +482,8 @@ github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJA
github.com/godbus/dbus v0.0.0-20190422162347-ade71ed3457e/go.mod h1:bBOAhwG1umN6/6ZUMtDFBMQR8jRg9O75tm9K00oMsK4=
github.com/gofrs/flock v0.0.0-20190320160742-5135e617513b/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
github.com/gofrs/flock v0.7.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
github.com/gogo/protobuf v1.0.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
Expand Down Expand Up @@ -662,6 +675,10 @@ github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtng
github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
github.com/hashicorp/go-immutable-radix v1.2.0 h1:l6UW37iCXwZkZoAbEYnptSHVE/cQ5bOTPYG5W3vf9+8=
github.com/hashicorp/go-immutable-radix v1.2.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
github.com/hashicorp/go-memdb v1.2.1 h1:wI9btDjYUOJJHTCnRlAG/TkRyD/ij7meJMrLK9X31Cc=
github.com/hashicorp/go-memdb v1.2.1/go.mod h1:OSvLJ662Jim8hMM+gWGyhktyWk2xPCnWMc7DWIqtkGA=
github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
github.com/hashicorp/go-multierror v0.0.0-20161216184304-ed905158d874/go.mod h1:JMRHfdO9jKNzS/+BTlxCjKNQHg/jZAft8U7LloJvN7I=
github.com/hashicorp/go-multierror v0.0.0-20171204182908-b7773ae21874/go.mod h1:JMRHfdO9jKNzS/+BTlxCjKNQHg/jZAft8U7LloJvN7I=
Expand All @@ -675,6 +692,7 @@ github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerX
github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-version v1.0.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
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=
Expand Down Expand Up @@ -1082,6 +1100,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
github.com/syndtr/gocapability v0.0.0-20170704070218-db04d3cc01c8/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
github.com/tdakkota/asciicheck v0.0.0-20200416190851-d7f85be797a2/go.mod h1:yHp0ai0Z9gUljN3o0xMhYJnH/IcvkdTBOX2fmJ93JEM=
Expand Down Expand Up @@ -1696,6 +1716,8 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d h1:LCPbGQ34PMrwad11aMZ+dbz5SAsq/0ySjRwQ8I9Qwd8=
gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
helm.sh/helm/v3 v3.1.1/go.mod h1:WYsFJuMASa/4XUqLyv54s0U/f3mlAaRErGmyy4z921g=
honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
Expand Down
8 changes: 1 addition & 7 deletions pkg/adapter/mtping/adapter.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ import (
corev1 "k8s.io/api/core/v1"

kubeclient "knative.dev/pkg/client/injection/kube/client"
"knative.dev/pkg/controller"
"knative.dev/pkg/logging"

"knative.dev/eventing/pkg/adapter/v2"
Expand Down Expand Up @@ -72,14 +71,9 @@ func NewAdapter(ctx context.Context, _ adapter.EnvConfigAccessor, ceClient cloud

// Start implements adapter.Adapter
func (a *mtpingAdapter) Start(ctx context.Context) error {
ctrl := NewController(ctx, a)

a.logger.Info("Starting controllers...")
go controller.StartAll(ctx, ctrl)

defer a.runner.Stop()
a.logger.Info("Starting job runner...")
a.runner.Start(ctx.Done())
defer a.runner.Stop()

a.logger.Infof("runner stopped")
return nil
Expand Down
8 changes: 5 additions & 3 deletions pkg/adapter/v2/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -179,18 +179,20 @@ func (e *EnvConfig) GetCloudEventOverrides() (*duckv1.CloudEventOverrides, error

func (e *EnvConfig) GetLeaderElectionConfig() (*kle.ComponentConfig, error) {
if e.LeaderElectionConfigJson == "" {
return defaultLeaderElectionConfig(), nil
return e.defaultLeaderElectionConfig(), nil
}

var config kle.ComponentConfig
if err := json.Unmarshal([]byte(e.LeaderElectionConfigJson), &config); err != nil {
return defaultLeaderElectionConfig(), err
return e.defaultLeaderElectionConfig(), err
}
config.Component = e.Component
return &config, nil
}

func defaultLeaderElectionConfig() *kle.ComponentConfig {
func (e *EnvConfig) defaultLeaderElectionConfig() *kle.ComponentConfig {
return &kle.ComponentConfig{
Component: e.Component,
Buckets: 1,
LeaseDuration: 15 * time.Second,
RenewDeadline: 10 * time.Second,
Expand Down
90 changes: 0 additions & 90 deletions pkg/adapter/v2/config_watcher.go

This file was deleted.

42 changes: 0 additions & 42 deletions pkg/adapter/v2/config_watcher_test.go

This file was deleted.

65 changes: 65 additions & 0 deletions pkg/adapter/v2/context.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
/*
Copyright 2020 The Knative Authors
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.
*/

package adapter

import (
"context"
)

type haEnabledKey struct{}

// WithHAEnabled signals to MainWithContext that it should set up an appropriate leader elector for this component.
func WithHAEnabled(ctx context.Context) context.Context {
return context.WithValue(ctx, haEnabledKey{}, struct{}{})
}

// IsHAEnabled checks the context for the desire to enable leader elector.
func IsHAEnabled(ctx context.Context) bool {
val := ctx.Value(haEnabledKey{})
return val != nil
}

type haDisabledFlagKey struct{}

// withHADisabledFlag signals to MainWithConfig that it should not set up an appropriate leader elector for this component.
func withHADisabledFlag(ctx context.Context) context.Context {
return context.WithValue(ctx, haDisabledFlagKey{}, struct{}{})
}

// isHADisabledFlag checks the context for the desired to disable leader elector.
func isHADisabledFlag(ctx context.Context) bool {
val := ctx.Value(haDisabledFlagKey{})
return val != nil
}

type controllerKey struct{}

// WithController signals to MainWithContext that it should
// create and configure a controller notifying the adapter
// when a resource is ready and removed
func WithController(ctx context.Context, ctor ControllerConstructor) context.Context {
return context.WithValue(ctx, controllerKey{}, ctor)
}

// ControllerFromContext gets the controller constructor from the context
func ControllerFromContext(ctx context.Context) ControllerConstructor {
value := ctx.Value(controllerKey{})
if value == nil {
return nil
}
return value.(ControllerConstructor)
}
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,23 @@ See the License for the specific language governing permissions and
limitations under the License.
*/

package leaderelection
package adapter

import (
kle "knative.dev/pkg/leaderelection"
"context"
"testing"

"knative.dev/pkg/controller"

_ "knative.dev/pkg/client/injection/kube/client/fake"
)

var ValidateConfig = kle.NewConfigFromConfigMap
func TestWithController(t *testing.T) {
ctx := WithController(context.TODO(), func(ctx context.Context, adapter Adapter) *controller.Impl {
return nil
})

if ControllerFromContext(ctx) == nil {
t.Error("expected non-nil controller constructor")
}
}
Loading

0 comments on commit adda82f

Please sign in to comment.