Remove Istio dependency from Eventing (Part - 1) #1044
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,49 +25,47 @@ import ( | |
| "fmt" | ||
| "log" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" | ||
| "github.com/knative/eventing/pkg/channelwatcher" | ||
| "github.com/knative/eventing/pkg/logging" | ||
| "github.com/knative/eventing/pkg/sidecar/fanout" | ||
| "github.com/knative/eventing/pkg/sidecar/multichannelfanout" | ||
| "github.com/knative/eventing/pkg/sidecar/swappable" | ||
| "go.uber.org/zap" | ||
| "go.uber.org/zap/zapcore" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/apimachinery/pkg/types" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| "sigs.k8s.io/controller-runtime/pkg/client/config" | ||
| "sigs.k8s.io/controller-runtime/pkg/manager" | ||
| "sigs.k8s.io/controller-runtime/pkg/runtime/signals" | ||
| // Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters). | ||
| // _ "k8s.io/client-go/plugin/pkg/client/auth/gcp" | ||
| ) | ||
|
|
||
| var ( | ||
| readTimeout = 1 * time.Minute | ||
| writeTimeout = 1 * time.Minute | ||
|
|
||
| port int | ||
| channelProvisioners listFlags | ||
| ) | ||
|
|
||
| type listFlags []string | ||
|
|
||
| func (l *listFlags) String() string { | ||
| return "" | ||
| } | ||
| func (l *listFlags) Set(value string) error { | ||
| *l = append(*l, value) | ||
| return nil | ||
| } | ||
|
|
||
| func init() { | ||
| flag.IntVar(&port, "sidecar_port", -1, "The port to run the sidecar on.") | ||
| flag.Var(&channelProvisioners, "channel_provisioner", "The provisioner of the channels that will be watched.") | ||
| } | ||
|
|
||
| func main() { | ||
|
|
@@ -84,14 +82,18 @@ func main() { | |
| logger.Fatal("--sidecar_port flag must be set") | ||
| } | ||
|
|
||
| if len(channelProvisioners) < 1 { | ||
| logger.Fatal("--channel_provisioner must be specified") | ||
| } | ||
|
|
||
| sh, err := swappable.NewEmptyHandler(logger) | ||
| if err != nil { | ||
| logger.Fatal("Unable to create swappable.Handler", zap.Error(err)) | ||
| } | ||
|
|
||
| mgr, err := setupChannelWatcher(logger, sh.UpdateConfig) | ||
|
akashrv marked this conversation as resolved.
|
||
| if err != nil { | ||
| logger.Fatal("Unable to create channel watcher.", zap.Error(err)) | ||
| } | ||
|
|
||
| s := &http.Server{ | ||
|
|
@@ -125,57 +127,87 @@ func main() { | |
| } | ||
| } | ||
|
|
||
| func setupChannelWatcher(logger *zap.Logger, configUpdated swappable.UpdateConfig) (manager.Manager, error) { | ||
| mgr, err := manager.New(config.GetConfigOrDie(), manager.Options{}) | ||
| if err != nil { | ||
| logger.Error("Error creating new maanger.", zap.Error(err)) | ||
| return nil, err | ||
| } | ||
| if err = v1alpha1.AddToScheme(mgr.GetScheme()); err != nil { | ||
| logger.Error("Error while adding eventing scheme to manager.", zap.Error(err)) | ||
|
Member
There was a problem hiding this comment. Should this include the scheme?
Contributor
Author
There was a problem hiding this comment. Is the question whether scheme is needed or should it be in this function?
|
||
| return nil, err | ||
| } | ||
| channelwatcher.New(mgr, logger, updateChannelConfig(configUpdated)) | ||
|
|
||
| return mgr, nil | ||
| } | ||
|
|
||
| func updateChannelConfig(updateConfig swappable.UpdateConfig) channelwatcher.WatchHandlerFunc { | ||
|
akashrv marked this conversation as resolved.
|
||
| return func(ctx context.Context, c client.Client, chanNamespacedName types.NamespacedName) error { | ||
| channels, err := listAllChannels(ctx, c) | ||
| if err != nil { | ||
| logging.FromContext(ctx).Info("Unable to list channels", zap.Error(err)) | ||
| return err | ||
| } | ||
| config := multiChannelFanoutConfig(channels) | ||
| return updateConfig(config) | ||
| } | ||
| } | ||
|
|
||
| func listAllChannels(ctx context.Context, c client.Client) ([]v1alpha1.Channel, error) { | ||
| channels := make([]v1alpha1.Channel, 0) | ||
| for { | ||
|
Member
There was a problem hiding this comment. It might be worth a comment here indicating that this is a do... while loop on
Member
There was a problem hiding this comment. In particular, I'm surprised that the A reading of the
Contributor
There was a problem hiding this comment. This is probably copied from other code that added pagination support defensively (@Harwayne may be able to provide context). Based on my reading of the original design for apiserver pagination, it appears @evankanderson is correct: pagination is opt-in by specifying the
Contributor
There was a problem hiding this comment. I am paranoid. I don't think it gets used today. I believe @akashrv looked into the Controller Runtime code in particular and confirmed that it isn't used. But the fact that it is in the interface makes me wary of leaving it out. I agree that ListOptions.Limit clearly talks about Continue with regards to having a limit set, but ListOptions.Continue doesn't specify that is the only time Continue will be used. I agree with your reading of the origin design that this is purely an opt-in feature and that not providing a limit means get everything (even if that is 500 MB). Having been shown this, I am OK removing it.
Contributor
Author
There was a problem hiding this comment. Sounds good. I'll simplify it and revert it to how it was in first iteration. |
||
| cl := &v1alpha1.ChannelList{} | ||
| opts := &client.ListOptions{ | ||
| // Set Raw because if we need to get more than one page, then we will put the continue token | ||
| // into opts.Raw.Continue. | ||
| Raw: &metav1.ListOptions{}, | ||
| } | ||
| if err := c.List(ctx, opts, cl); err != nil { | ||
| return nil, err | ||
| } | ||
| for _, c := range cl.Items { | ||
| if c.Status.IsReady() && shouldWatch(&c) { | ||
| channels = append(channels, c) | ||
| } | ||
| } | ||
| if cl.Continue != "" { | ||
| opts.Raw.Continue = cl.Continue | ||
| } else { | ||
| return channels, nil | ||
| } | ||
|
akashrv marked this conversation as resolved.
|
||
| } | ||
| } | ||
|
|
||
| func shouldWatch(ch *v1alpha1.Channel) bool { | ||
| if ch.Spec.Provisioner != nil && ch.Spec.Provisioner.Namespace == "" { | ||
|
Member
There was a problem hiding this comment. Maybe make namespace non-empty an early exit with a comment that we only support cluster-scoped provisioners (so it's easier to extend to namespace-scoped provisioners later)?
Contributor
Author
There was a problem hiding this comment. Didn't understand the comment. This function is used to filter out channels in the handler inside the watch. It is not used in the request path.
Member
There was a problem hiding this comment. I was suggesting writing this as: if ch.Spec.Provisioner == nil || ch.Spec.Provisioner.Namespace != "" {
// Only support cluster-level provisioners right now.
return false
}
... |
||
| for _, v := range channelProvisioners { | ||
|
Member
There was a problem hiding this comment. Would it make more sense to make
Contributor
Author
There was a problem hiding this comment. We don't expect to have too many provisioners. Most cases this should be just one check without a for loop. |
||
| if v == ch.Spec.Provisioner.Name { | ||
| return true | ||
| } | ||
| } | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func multiChannelFanoutConfig(channels []v1alpha1.Channel) *multichannelfanout.Config { | ||
| cc := make([]multichannelfanout.ChannelConfig, 0) | ||
| for _, c := range channels { | ||
| channelConfig := multichannelfanout.ChannelConfig{ | ||
| Namespace: c.Namespace, | ||
| Name: c.Name, | ||
| HostName: c.Status.Address.Hostname, | ||
| } | ||
| if c.Spec.Subscribable != nil { | ||
| channelConfig.FanoutConfig = fanout.Config{ | ||
| Subscriptions: c.Spec.Subscribable.Subscribers, | ||
| } | ||
| } | ||
| cc = append(cc, channelConfig) | ||
| } | ||
| return &multichannelfanout.Config{ | ||
| ChannelConfigs: cc, | ||
| } | ||
| } | ||
|
|
||
| // runnableServer is a small wrapper around http.Server so that it matches the manager.Runnable | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This doesn't seem right -- why not just remove it?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I kept it around as it helps with debugging. When I started working on controllers or anything that connects to GKE from my local machine I would get some config related error. After searching online I found threads that it is because of missing package. So I added it here and then commented it out as a nicety that could help others working on the code base.