cronjobsource replaced with pingsource, dependencies and tests updated

Author: tzununbekov

cloudbuild.yaml

@@ -11,6 +11,8 @@ steps:
   - ${_KANIKO_EXTRA_ARGS}
   waitFor: ['-']
 
+timeout: 1800s
+
 substitutions:
   _KANIKO_IMAGE_TAG: "latest"
   _KANIKO_NO_PUSH: "false"

go.mod

@@ -3,41 +3,35 @@ module github.com/triggermesh/tm
 go 1.13
 
 require (
-	contrib.go.opencensus.io/exporter/ocagent v0.6.0 // indirect
-	contrib.go.opencensus.io/exporter/stackdriver v0.12.9 // indirect
 	github.com/docker/spdystream v0.0.0-20181023171402-6480d4af844c // indirect
-	github.com/dsnet/compress v0.0.1 // indirect
-	github.com/frankban/quicktest v1.7.2 // indirect
 	github.com/ghodss/yaml v1.0.0
-	github.com/golang/groupcache v0.0.0-20191027212112-611e8accdfc9 // indirect
+	github.com/json-iterator/go v1.1.10 // indirect
 	github.com/knative/build v0.7.0
 	github.com/knative/pkg v0.0.0-20190624141606-d82505e6c5b4
 	github.com/mholt/archiver v2.1.0+incompatible
-	github.com/nwaples/rardecode v1.0.0 // indirect
 	github.com/olekukonko/tablewriter v0.0.4
-	github.com/openzipkin/zipkin-go v0.2.2 // indirect
 	github.com/pierrec/lz4 v2.4.0+incompatible // indirect
-	github.com/robfig/cron v1.2.0 // indirect
-	github.com/sirupsen/logrus v1.4.2
+	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/afero v1.2.2
-	github.com/spf13/cobra v0.0.5
-	github.com/stretchr/testify v1.4.0
-	github.com/tektoncd/pipeline v0.11.2
+	github.com/spf13/cobra v1.0.0
+	github.com/stretchr/testify v1.5.1
+	github.com/tektoncd/pipeline v0.14.0
 	github.com/tektoncd/triggers v0.1.0
 	github.com/tidwall/gjson v1.3.2 // indirect
-	golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413
-	google.golang.org/genproto v0.0.0-20191028173616-919d9bdd9fe6 // indirect
+	golang.org/x/crypto v0.0.0-20200707235045-ab33eee955e0
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect
+	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
 	gopkg.in/src-d/go-git.v4 v4.13.1
-	gopkg.in/yaml.v2 v2.2.5
-	k8s.io/api v0.17.2
-	k8s.io/apimachinery v0.17.2
-	k8s.io/client-go v0.17.0
-	k8s.io/utils v0.0.0-20191030222137-2b95a09bc58d // indirect
-	knative.dev/eventing v0.12.0
-	knative.dev/eventing-contrib v0.11.2
-	knative.dev/pkg v0.0.0-20200207155214-fef852970f43
-	knative.dev/serving v0.12.0
-	sigs.k8s.io/controller-runtime v0.4.0 // indirect
+	gopkg.in/yaml.v2 v2.3.0
+	k8s.io/api v0.18.5
+	k8s.io/apimachinery v0.18.5
+	k8s.io/client-go v12.0.0+incompatible
+	k8s.io/klog/v2 v2.3.0 // indirect
+	k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19 // indirect
+	knative.dev/eventing v0.15.0
+	knative.dev/eventing-contrib v0.15.0
+	knative.dev/pkg v0.0.0-20200708171447-5358179e7499
+	knative.dev/serving v0.15.0
 )
 
 replace (
@@ -61,5 +55,4 @@ replace (
 	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.0.0-20191016115707-22244e5b01eb
 	k8s.io/metrics => k8s.io/metrics v0.0.0-20191016113728-f445c7b35c1c
 	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.0.0-20191016112728-ceb381866e80
-	knative.dev/pkg => knative.dev/pkg v0.0.0-20200122022923-4e81bc3c320f
 )

go.sum

@@ -32,7 +32,6 @@ import (
 	printerwrapper "github.com/triggermesh/tm/pkg/printer"
 	githubSource "knative.dev/eventing-contrib/github/pkg/client/clientset/versioned"
 	eventingApi "knative.dev/eventing/pkg/client/clientset/versioned"
-	legacyEventingApi "knative.dev/eventing/pkg/legacyclient/clientset/versioned"
 	servingApi "knative.dev/serving/pkg/client/clientset/versioned"
 
 	"k8s.io/client-go/kubernetes"
@@ -52,11 +51,6 @@ const (
 var (
 	// Namespace to work in passed with "-n" argument or defined in kube configs
 	Namespace string
-	// Registry to store docker images for user services
-	// Default value for tm cloud is knative.registry.svc.cluster.local
-	// RegistryHost    string
-	// RegistrySecret  string
-	// RegistrySkipTLS bool
 	// Output format for k8s objects in "tm get" result. Can be either "yaml" (default) or "json"
 	Output string
 	// Debug enables verbose output for CLI commands
@@ -67,6 +61,7 @@ var (
 	Wait bool
 )
 
+// Registry to store docker images for user services
 type Registry struct {
 	Host    string
 	Secret  string
@@ -78,7 +73,6 @@ type ConfigSet struct {
 	Core            *kubernetes.Clientset
 	Build           *buildApi.Clientset
 	Serving         *servingApi.Clientset
-	LegacyEventing  *legacyEventingApi.Clientset
 	Eventing        *eventingApi.Clientset
 	GithubSource    *githubSource.Clientset
 	TektonPipelines *tektonResource.Clientset
@@ -185,9 +179,6 @@ func NewClient(cfgFile string, output ...io.Writer) (ConfigSet, error) {
 	if c.Eventing, err = eventingApi.NewForConfig(config); err != nil {
 		return c, err
 	}
-	if c.LegacyEventing, err = legacyEventingApi.NewForConfig(config); err != nil {
-		return c, err
-	}
 	if c.Build, err = buildApi.NewForConfig(config); err != nil {
 		return c, err
 	}

pkg/client/client.go

@@ -38,26 +38,27 @@ type TriggermeshProvider struct {
 
 // Function describes function definition in serverless format
 type Function struct {
-	Handler     string                   `yaml:"handler,omitempty"`
-	Source      string                   `yaml:"source,omitempty"`
-	Revision    string                   `yaml:"revision,omitempty"`
-	Runtime     string                   `yaml:"runtime,omitempty"`
-	Concurrency int                      `yaml:"concurrency,omitempty"`
-	Buildargs   []string                 `yaml:"buildargs,omitempty"`
-	Description string                   `yaml:"description,omitempty"`
-	Labels      []string                 `yaml:"labels,omitempty"`
-	Environment map[string]string        `yaml:"environment,omitempty"`
-	EnvSecrets  []string                 `yaml:"env-secrets,omitempty"`
-	Annotations map[string]string        `yaml:"annotations,omitempty"`
-	Events      []map[string]interface{} `yaml:"events,omitempty"`
+	Handler     string            `yaml:"handler,omitempty"`
+	Source      string            `yaml:"source,omitempty"`
+	Revision    string            `yaml:"revision,omitempty"`
+	Runtime     string            `yaml:"runtime,omitempty"`
+	Concurrency int               `yaml:"concurrency,omitempty"`
+	Buildargs   []string          `yaml:"buildargs,omitempty"`
+	Description string            `yaml:"description,omitempty"`
+	Labels      []string          `yaml:"labels,omitempty"`
+	Environment map[string]string `yaml:"environment,omitempty"`
+	EnvSecrets  []string          `yaml:"env-secrets,omitempty"`
+	Annotations map[string]string `yaml:"annotations,omitempty"`
+	Schedule    []Schedule        `yaml:"schedule,omitempty"`
 }
 
-// Schedule represents simple structure of event schedule
-// with cronjob-style rate string and data to use in event.
-// Deprecated.
+// Schedule struct contains a data in JSON format and a cron
+// that defines how often events should be sent to a function.
+// Description string may be used to explain events purpose.
 type Schedule struct {
-	Rate string
-	Data string
+	Cron        string
+	JSONData    string
+	Description string
 }
 
 // Aos returns filesystem object with standard set of os methods implemented by afero package

pkg/file/serverless.go

@@ -20,7 +20,10 @@ func NewLogger() *StandardLogger {
 	var baseLogger = logrus.New()
 	var standardLogger = &StandardLogger{baseLogger}
 
-	standardLogger.Formatter = &logrus.TextFormatter{}
+	standardLogger.Formatter = &logrus.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: "15:04:05",
+	}
 
 	return standardLogger
 }

pkg/log/log.go

@@ -29,8 +29,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	legacysourcesv1alpha1 "knative.dev/eventing/pkg/apis/legacysources/v1alpha1"
-	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
+	sourcesv1alpha2 "knative.dev/eventing/pkg/apis/sources/v1alpha2"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 // Push tries to read git configuration in current directory and if it succeeds
@@ -137,8 +137,8 @@ func Push(clientset *client.ConfigSet, token string) error {
 	return nil
 }
 
-func getContainerSource(project, owner, token string) *legacysourcesv1alpha1.ContainerSource {
-	return &legacysourcesv1alpha1.ContainerSource{
+func getContainerSource(project, owner, token string) *sourcesv1alpha2.ContainerSource {
+	return &sourcesv1alpha2.ContainerSource{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "ContainerSource",
 			APIVersion: "sources.eventing.knative.dev/v1alpha1",
@@ -147,15 +147,17 @@ func getContainerSource(project, owner, token string) *legacysourcesv1alpha1.Con
 			Name:      project,
 			Namespace: client.Namespace,
 		},
-		Spec: legacysourcesv1alpha1.ContainerSourceSpec{
-			Sink: &duckv1beta1.Destination{
-				Ref: &corev1.ObjectReference{
-					Kind:       "Service",
-					APIVersion: "serving.knative.dev/v1beta1",
-					Name:       project + "-transceiver",
+		Spec: sourcesv1alpha2.ContainerSourceSpec{
+			SourceSpec: duckv1.SourceSpec{
+				Sink: duckv1.Destination{
+					Ref: &duckv1.KReference{
+						Kind:       "Service",
+						APIVersion: "serving.knative.dev/v1beta1",
+						Name:       project + "-transceiver",
+					},
 				},
 			},
-			Template: &corev1.PodTemplateSpec{
+			Template: corev1.PodTemplateSpec{
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
 						{
@@ -269,14 +271,14 @@ func createOrUpdateConfigmap(clientset *client.ConfigSet, cm *corev1.ConfigMap)
 	return nil
 }
 
-func createOrUpdateContainersource(clientset *client.ConfigSet, cs *legacysourcesv1alpha1.ContainerSource) error {
-	if _, err := clientset.LegacyEventing.SourcesV1alpha1().ContainerSources(cs.Namespace).Create(cs); k8sErrors.IsAlreadyExists(err) {
-		csObj, err := clientset.LegacyEventing.SourcesV1alpha1().ContainerSources(cs.Namespace).Get(cs.Name, metav1.GetOptions{})
+func createOrUpdateContainersource(clientset *client.ConfigSet, cs *sourcesv1alpha2.ContainerSource) error {
+	if _, err := clientset.Eventing.SourcesV1alpha2().ContainerSources(cs.Namespace).Create(cs); k8sErrors.IsAlreadyExists(err) {
+		csObj, err := clientset.Eventing.SourcesV1alpha2().ContainerSources(cs.Namespace).Get(cs.Name, metav1.GetOptions{})
 		if err != nil {
 			return err
 		}
 		cs.ObjectMeta.ResourceVersion = csObj.GetResourceVersion()
-		if _, err := clientset.LegacyEventing.SourcesV1alpha1().ContainerSources(cs.Namespace).Update(cs); err != nil {
+		if _, err := clientset.Eventing.SourcesV1alpha2().ContainerSources(cs.Namespace).Update(cs); err != nil {
 			return err
 		}
 	} else if err != nil {

pkg/push/push.go

@@ -22,14 +22,18 @@ import (
 	"time"
 
 	"github.com/ghodss/yaml"
-	"github.com/triggermesh/tm/pkg/client"
 	corev1 "k8s.io/api/core/v1"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"knative.dev/pkg/apis"
 	servingv1 "knative.dev/serving/pkg/apis/serving/v1"
+
+	"github.com/triggermesh/tm/pkg/client"
 )
 
+const defaultBuildTimeout = 10
+
 // Deploy receives Service structure and generate knative/service object to deploy it in knative cluster
 func (s *Service) Deploy(clientset *client.ConfigSet) (string, error) {
 	var err error
@@ -119,12 +123,20 @@ func (s *Service) Deploy(clientset *client.ConfigSet) (string, error) {
 		return "", fmt.Errorf("Creating service: %s", err)
 	}
 
-	// TODO Add cronjob yaml into --dry output
-	// if len(s.Cronjob.Schedule) != 0 {
-	// 	if err := s.CreateCronjobSource(clientset); err != nil {
-	// 		return "", fmt.Errorf("Creating cronjob source: %s", err)
-	// 	}
-	// }
+	// before creating PingSources remove old ones
+	// to make sure that we're in sync with manifest
+	if err := s.removePingSources(service.UID, clientset); err != nil {
+		clientset.Log.Warnf("Failed to remove schedule: %v\n", err)
+	}
+
+	for _, sched := range s.Schedule {
+		ps := s.pingSource(sched.Cron, sched.JSONData, service.UID)
+		clientset.Log.Infof("Creating %q schedule\n", ps.Spec.Schedule)
+		err := s.createPingSource(ps, clientset)
+		if err != nil {
+			clientset.Log.Errorf("Failed to create schedule: %v\n", err)
+		}
+	}
 
 	if !client.Wait {
 		return fmt.Sprintf("Deployment started. Run \"tm -n %s describe service %s\" to see details", s.Namespace, s.Name), nil
@@ -210,7 +222,7 @@ func (s *Service) wait(clientset *client.ConfigSet) (string, error) {
 
 	duration, err := time.ParseDuration(s.BuildTimeout)
 	if err != nil {
-		duration = 10 * time.Minute
+		duration = defaultBuildTimeout * time.Minute
 	}
 
 	ticker := time.NewTicker(duration)
@@ -265,7 +277,7 @@ func (s *Service) wait(clientset *client.ConfigSet) (string, error) {
 				}
 			}
 		case <-ticker.C:
-			return "", fmt.Errorf("watch service timeout")
+			return "", fmt.Errorf("Service %q didn't become ready in time", s.Name)
 		}
 	}
 }