Cel filters -- Review

Gopkg.toml

@@ -10,7 +10,6 @@ required = [
   "k8s.io/code-generator/cmd/informer-gen",
   "github.com/knative/test-infra/scripts",
   "github.com/knative/test-infra/tools/dep-collector",
-  "github.com/knative/build/pkg/apis/build/v1alpha1",
 ]
 
 [prune]
@@ -27,56 +26,40 @@ required = [
   name = "github.com/knative/test-infra"
   non-go = false
 
-# Use HEAD (2018-04-21) to pick up:
-# https://github.com/spf13/cobra/pull/662
-# TODO this can be changed to a version constraint on 0.0.3 which has this commit
-# [[constraint]]
-#   name = "github.com/spf13/cobra"
-#   version = "0.0.3"
-[[override]]
-  name = "github.com/spf13/cobra"
-  revision = "615425954c3b0d9485a7027d4d451fdcdfdee84e"
-
-# TODO this can be changed to a version constraint on 1.0.3 which has this commit
-# [[constraint]]
-#   name = "github.com/spf13/pflag"
-#   version = "1.0.3"
-[[override]]
-  name = "github.com/spf13/pflag"
-  revision = "583c0c0531f06d5278b7d917446061adc344b5cd"
-
 # TODO why is this overridden?
 [[override]]
   name = "gopkg.in/yaml.v2"
   version = "v2.2.1"
-
+# TODO(https://github.com/knative/eventing/issues/1065): if we get to update k8s.io to > 1.13, we can remove
+# k8s-dynamic-fake-simple.patch in ./hack/update-deps.sh
+#
 # Overridden to ensure compatibility with GKE
 # GKE version as of 2019-01-24 is 1.11
 # controller-runtime 0.1.9 requires at least 1.12
 [[override]]
   name = "k8s.io/api"
-  version = "kubernetes-1.12.5"
+  version = "kubernetes-1.12.6"
 
 # Overridden to ensure compatibility with GKE
 # GKE version as of 2019-01-24 is 1.11
 # controller-runtime 0.1.9 requires at least 1.12
 [[override]]
   name = "k8s.io/apimachinery"
-  version = "kubernetes-1.12.5"
+  version = "kubernetes-1.12.6"
 
 # Overridden to ensure compatibility with GKE
 # GKE version as of 2019-01-24 is 1.11
 # controller-runtime 0.1.9 requires at least 1.12
 [[override]]
   name = "k8s.io/code-generator"
-  version = "kubernetes-1.12.5"
+  version = "kubernetes-1.12.6"
 
 # Overridden to ensure compatibility with GKE
 # GKE version as of 2019-01-24 is 1.11
 # controller-runtime 0.1.9 requires at least 1.12
 [[override]]
   name = "k8s.io/client-go"
-  version = "kubernetes-1.12.5"
+  version = "kubernetes-1.12.6"
 
 # This is the commit at which k8s depends on this in 1.11
 # It seems to be broken at HEAD.
@@ -86,22 +69,12 @@ required = [
   name = "github.com/json-iterator/go"
   revision = "f2b4162afba35581b6d4a50d3b8f34e33c144682"
 
-# TODO why is this overridden?
-[[override]]
-  name = "github.com/golang/protobuf"
-  version = "v1.1.0"
-
-# TODO why is this overridden?
-[[override]]
-  name = "golang.org/x/oauth2"
-  revision = "cdc340f7c179dbbfa4afd43b7614e8fcadde4269"
-
 # Constrain the version of knative/pkg we would like to import.
 # This controls when we upgrade apis independently of Serving.
-[[constraint]]
+[[override]]
   name = "github.com/knative/pkg"
-  # HEAD as of 2019-04-08
-  revision = "2b574edcd712e848556c69cc95a2622145284882"
+  # HEAD as of 2019-05-01
+  revision = "f95c27ee8cdceb4545b57c7d4cf283a94d303dca"
 
 # TODO why is this overridden?
 [[override]]
@@ -130,4 +103,13 @@ required = [
 
 [[constraint]]
   name = "github.com/cloudevents/sdk-go"
-  version = "=0.4.4"
+  version = "=0.6.0"
+
+[[constraint]]
+  name = "github.com/google/cel-go"
+  version = "=0.2.0"
+
+# needed because pkg upgraded
+[[override]]
+  name = "go.uber.org/zap"
+  revision = "67bc79d13d155c02fd008f721863ff8cc5f30659"

OWNERS_ALIASES

@@ -0,0 +1,21 @@
+aliases:
+  # These aliases are for OWNERS of the various Channel implementations. These
+  # Are in addition to the repo level OWNERS.
+
+  gcp-pubsub-approvers:
+    - Harwayne
+  gcp-pubsub-reviewers:
+    - Harwayne
+
+  kafka-approvers:
+    - bbrowning
+    - matzew
+  kafka-reviewers:
+    - bbrowning
+    - matzew
+
+  natss-approvers:
+    - Abd4llA
+  natss-reviewers:
+    - Abd4llA
+

cmd/apiserver_receive_adapter/main.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2019 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 main
+
+import (
+	"flag"
+	"time"
+
+	"k8s.io/client-go/rest"
+
+	// 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"
+
+	"github.com/knative/eventing/pkg/reconciler"
+
+	"github.com/kelseyhightower/envconfig"
+	"github.com/knative/eventing/pkg/adapter/apiserver"
+	"github.com/knative/eventing/pkg/kncloudevents"
+	"github.com/knative/pkg/apis/duck"
+	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
+	kncontroller "github.com/knative/pkg/controller"
+	"github.com/knative/pkg/signals"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+var (
+	masterURL  = flag.String("master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+	kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
+)
+
+type envConfig struct {
+	SinkURI    string   `split_words:"true" required:"true"`
+	ApiVersion []string `split_words:"true" required:"true"`
+	Kind       []string `required:"true"`
+	Controller []bool   `required:"true"`
+}
+
+func main() {
+	flag.Parse()
+
+	logCfg := zap.NewProductionConfig()
+	logCfg.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
+	dlogger, err := logCfg.Build()
+	logger := dlogger.Sugar()
+
+	var env envConfig
+	err = envconfig.Process("", &env)
+	if err != nil {
+		logger.Fatalw("Error processing environment", zap.Error(err))
+	}
+
+	// set up signals so we handle the first shutdown signal gracefully
+	stopCh := signals.SetupSignalHandler()
+
+	cfg, err := clientcmd.BuildConfigFromFlags(*masterURL, *kubeconfig)
+	if err != nil {
+		logger.Fatalw("Error building kubeconfig", zap.Error(err))
+	}
+
+	logger = logger.With(zap.String("controller/apiserver", "adapter"))
+	logger.Info("Starting the controller")
+
+	numControllers := len(env.ApiVersion)
+	cfg.QPS = float32(numControllers) * rest.DefaultQPS
+	cfg.Burst = numControllers * rest.DefaultBurst
+	opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh)
+
+	client, err := dynamic.NewForConfig(cfg)
+	if err != nil {
+		logger.Fatalw("Error building dynamic client", zap.Error(err))
+	}
+
+	eventsClient, err := kncloudevents.NewDefaultClient(env.SinkURI)
+	if err != nil {
+		logger.Fatalw("Error building cloud event client", zap.Error(err))
+	}
+
+	controllers := []*kncontroller.Impl{}
+
+	// Create one controller per resource.
+	for i, apiVersion := range env.ApiVersion {
+		kind := env.Kind[i]
+		controlled := env.Controller[i]
+
+		obj := &duckv1alpha1.AddressableType{}
+
+		factory := duck.TypedInformerFactory{
+			Client:       client,
+			ResyncPeriod: time.Duration(10), // TODO
+			StopChannel:  stopCh,
+			Type:         obj,
+		}
+
+		gv, err := schema.ParseGroupVersion(apiVersion)
+		if err != nil {
+			logger.Fatalw("Error parsing APIVersion", zap.Error(err))
+		}
+
+		gvk := schema.GroupVersionKind{Kind: kind, Group: gv.Group, Version: gv.Version}
+
+		// This is really bad.
+		gvr, _ := meta.UnsafeGuessKindToResource(gvk)
+
+		// Get and start the informer for gvr
+		logger.Infof("Starting informer for %v", gvk)
+		informer, lister, err := factory.Get(gvr)
+		if err != nil {
+			logger.Fatalw("Error starting informer", zap.Error(err))
+		}
+		controllers = append(controllers, apiserver.NewController(opt, informer, lister, eventsClient, controlled))
+	}
+
+	// Start all of the controllers.
+	logger.Info("Starting controllers.")
+	go kncontroller.StartAll(stopCh, controllers...)
+	<-stopCh
+}

cmd/broker/ingress/main.go

@@ -29,9 +29,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cloudevents/sdk-go/pkg/cloudevents"
-	ceclient "github.com/cloudevents/sdk-go/pkg/cloudevents/client"
-	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+	cloudevents "github.com/cloudevents/sdk-go"
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	"github.com/knative/eventing/pkg/broker"
 	"github.com/knative/eventing/pkg/provisioners"
@@ -42,10 +40,11 @@ import (
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 	"go.uber.org/zap"
-	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	crlog "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	// 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 (
@@ -85,7 +84,7 @@ func main() {
 		Path:   "/",
 	}
 
-	ceClient, err := ceclient.NewDefault()
+	ceClient, err := cloudevents.NewDefaultClient()
 	if err != nil {
 		logger.Fatal("Unable to create CE client", zap.Error(err))
 	}
@@ -157,7 +156,7 @@ func getRequiredEnv(envKey string) string {
 
 type handler struct {
 	logger     *zap.Logger
-	ceClient   ceclient.Client
+	ceClient   cloudevents.Client
 	channelURI *url.URL
 	brokerName string
 }
@@ -191,7 +190,7 @@ func (h *handler) Start(stopCh <-chan struct{}) error {
 }
 
 func (h *handler) serveHTTP(ctx context.Context, event cloudevents.Event, resp *cloudevents.EventResponse) error {
-	tctx := cehttp.TransportContextFrom(ctx)
+	tctx := cloudevents.HTTPTransportContextFrom(ctx)
 	if tctx.Method != http.MethodPost {
 		resp.Status = http.StatusMethodNotAllowed
 		return nil
@@ -220,7 +219,7 @@ func (h *handler) serveHTTP(ctx context.Context, event cloudevents.Event, resp *
 	return h.sendEvent(ctx, tctx, event)
 }
 
-func (h *handler) sendEvent(ctx context.Context, tctx cehttp.TransportContext, event cloudevents.Event) error {
+func (h *handler) sendEvent(ctx context.Context, tctx cloudevents.HTTPTransportContext, event cloudevents.Event) error {
 	sendingCTX := broker.SendingContext(ctx, tctx, h.channelURI)
 
 	startTS := time.Now()
@@ -246,7 +245,11 @@ func (h *handler) decrementTTL(event *cloudevents.Event) bool {
 		return false
 	}
 
-	event.Context = broker.SetTTL(event.Context, ttl)
+	var err error
+	event.Context, err = broker.SetTTL(event.Context, ttl)
+	if err != nil {
+		h.logger.Error("failed to set TTL", zap.Error(err))
+	}
 	return true
 }