added valImage to monitor image vulnerabilities

tools/monitoring/alert/alert.go

@@ -29,21 +29,21 @@ import (
 	"github.com/knative/test-infra/tools/monitoring/subscriber"
 )
 
+const subName = "test-infra-monitoring-sub"
+
+var alertEmailRecipients = []string{"[email protected]"}
+
 // Client holds all the resources required to run alerting
 type Client struct {
 	*subscriber.Client
-	*MailConfig
-	db *mysql.DB
-}
-
-type MailConfig struct {
 	*mail.Config
-	Recipients []string
+	db *mysql.DB
 }
 
 // Setup sets up the client required to run alerting workflow
-func Setup(psClient *subscriber.Client, db *mysql.DB, mc *MailConfig) *Client {
-	return &Client{psClient, mc, db}
+func Setup(db *mysql.DB, mc *mail.Config) (*Client, error) {
+	sub, err := subscriber.NewSubscriberClient(subName)
+	return &Client{sub, mc, db}, err
 }
 
 // RunAlerting start the alerting workflow
@@ -120,13 +120,8 @@ func (c *Client) handleSingleError(config *config.Config, rmsg *prowapi.ReportMe
 	}
 }
 
-func (m *MailConfig) sendAlert(c *mailContent) error {
-	log.Printf("sending alert...")
-	return m.Send(m.Recipients, c.subject(), c.body())
-}
-
 // Alert checks alert condition and alerts table and send alert mail conditionally
-func (m *MailConfig) Alert(errorPattern string, s *config.SelectedConfig, db *mysql.DB) (bool, error) {
+func (c *Client) Alert(errorPattern string, s *config.SelectedConfig, db *mysql.DB) (bool, error) {
 	log.Println("Fetching error logs")
 	errorLogs, err := db.ListErrorLogs(errorPattern, s.Duration())
 	if err != nil {
@@ -151,7 +146,7 @@ func (m *MailConfig) Alert(errorPattern string, s *config.SelectedConfig, db *my
 	}
 
 	log.Println("Generating and sending the alert email")
-	content := mailContent{*report, errorPattern, s.Hint, s.Duration()}
-	err = m.sendAlert(&content)
+	mcont := mailContent{*report, errorPattern, s.Hint, s.Duration()}
+	err = c.Send(alertEmailRecipients, mcont.subject(), mcont.body())
 	return err == nil, err
 }

tools/monitoring/main.go

@@ -30,21 +30,15 @@ import (
 	"github.com/knative/test-infra/tools/monitoring/mail"
 	msql "github.com/knative/test-infra/tools/monitoring/mysql"
 	"github.com/knative/test-infra/tools/monitoring/subscriber"
+	"github.com/knative/test-infra/tools/monitoring/valimage"
 )
 
 var (
-	dbConfig   *mysql.DBConfig
-	mailConfig *mail.Config
-	client     *subscriber.Client
-	wfClient   *alert.Client
-	db         *msql.DB
-
-	alertEmailRecipients = []string{"[email protected]"}
-)
-
-const (
-	projectID = "knative-tests"
-	subName   = "test-infra-monitoring-sub"
+	dbConfig    *mysql.DBConfig
+	mailConfig  *mail.Config
+	client      *subscriber.Client
+	alertClient *alert.Client
+	db          *msql.DB
 )
 
 func main() {
@@ -78,20 +72,22 @@ func main() {
 		log.Fatal(err)
 	}
 
-	ctx := context.Background()
-	client, err = subscriber.NewSubscriberClient(ctx, projectID, subName)
-	if err != nil {
-		log.Fatalf("Failed to initialize the subscriber %+v", err)
-	}
-
 	err = gcs.Authenticate(context.Background(), *serviceAccount)
 	if err != nil {
 		log.Fatalf("Failed to authenticate gcs %+v", err)
 	}
 
-	wfClient = alert.Setup(client, db, &alert.MailConfig{Config: mailConfig, Recipients: alertEmailRecipients})
+	alertClient, err = alert.Setup(db, mailConfig)
+	if err != nil {
+		log.Fatalf("Failed to setup test-infra monitoring: %v\n", err)
+	}
+	alertClient.RunAlerting()
 
-	wfClient.RunAlerting()
+	imageClient, err := valimage.Setup(mailConfig)
+	if err != nil {
+		log.Fatalf("Failed to set up image vulernabilties monitoring. Error: %v\n", err)
+	}
+	imageClient.Run()
 
 	// use PORT environment variable, or default to 8080
 	port := "8080"
@@ -126,8 +122,14 @@ func sendTestEmail(w http.ResponseWriter, r *http.Request) {
 	log.Printf("Serving request: %s", r.URL.Path)
 	log.Println("Sending test email")
 
+	recipients, ok := r.URL.Query()["recipient"]
+	if !ok || len(recipients[0]) < 1 {
+		fmt.Fprintln(w, "Url Param 'recipient' is missing")
+		return
+	}
+
 	err := mailConfig.Send(
-		alertEmailRecipients,
+		recipients,
 		"Test Subject",
 		"Test Content",
 	)

tools/monitoring/subscriber/subscriber.go

@@ -25,6 +25,11 @@ import (
 	"github.com/knative/test-infra/tools/monitoring/prowapi"
 )
 
+const projectID = "knative-tests"
+
+// pubsub.Client is scoped to a single GCP project. Reuse the pubsub.Client as needed.
+var pubsubClient *pubsub.Client
+
 // Client is a wrapper on the subscriber Operation
 type Client struct {
 	Operation
@@ -37,13 +42,15 @@ type Operation interface {
 }
 
 // NewSubscriberClient returns a new SubscriberClient used to read crier pubsub messages
-func NewSubscriberClient(ctx context.Context, projectID string, subName string) (*Client, error) {
-	c, err := pubsub.NewClient(ctx, projectID)
-	if err != nil {
-		return nil, err
+func NewSubscriberClient(subName string) (*Client, error) {
+	var err error
+	if pubsubClient == nil {
+		log.Println("pubsub.Client not created yet. Creating the client.")
+		if pubsubClient, err = pubsub.NewClient(context.Background(), projectID); err != nil {
+			return nil, err
+		}
 	}
-
-	return &Client{c.Subscription(subName)}, nil
+	return &Client{pubsubClient.Subscription(subName)}, nil
 }
 
 // ReceiveMessageAckAll acknowledges all incoming pusub messages and convert the pubsub message to ReportMessage.

tools/monitoring/valimage/valImage.go

@@ -0,0 +1,92 @@
+// 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 valimage
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"cloud.google.com/go/pubsub"
+
+	"github.com/knative/test-infra/tools/monitoring/mail"
+	"github.com/knative/test-infra/tools/monitoring/subscriber"
+)
+
+var (
+	// monitoringSubs is a list of subscriptions to subscribe for image vulnerabilities
+	monitoringSubs = [...]string{
+		"sub-container-analysis-notes-v1beta1",
+		"sub-container-analysis-occurrences-v1beta1",
+	}
+	recipients = []string{"[email protected]"}
+)
+
+// Client holds resources for monitoring image vulnerabilities
+type Client struct {
+	subClients []*subscriber.Client
+	mailClient *mail.Config
+}
+
+// Setup initialize all the resources for monitoring image vulnerabilities
+func Setup(mconfig *mail.Config) (*Client, error) {
+	var subClients []*subscriber.Client
+	for _, sub := range monitoringSubs {
+		sub, err := subscriber.NewSubscriberClient(sub)
+		if err != nil {
+			return nil, err
+		}
+		subClients = append(subClients, sub)
+	}
+
+	return &Client{
+		subClients: subClients,
+		mailClient: mconfig,
+	}, nil
+}
+
+// Run start a background process that listens to the message
+func (c *Client) Run() {
+	log.Println("Starting image vulnerabilities monitoring")
+	for _, sub := range c.subClients {
+		go func() {
+			err := sub.Receive(context.Background(), func(ctx context.Context, msg *pubsub.Message) {
+				c.sendMessage(msg)
+				msg.Ack()
+			})
+			if err != nil {
+				log.Printf("Failed to receive messages due to: %v\n", err)
+			}
+		}()
+	}
+}
+
+func (c *Client) sendMessage(msg *pubsub.Message) {
+	err := c.mailClient.Send(recipients, "Image Vulnerabilities Detected", toMailContent(msg))
+	if err != nil {
+		log.Printf("Failed to send alert message %v\n", err)
+	}
+}
+
+func toMailContent(msg *pubsub.Message) string {
+	c := fmt.Sprintf("Message Data: %v\n", string(msg.Data))
+	if b, err := json.MarshalIndent(msg, "", "\t"); err == nil {
+		c += fmt.Sprintf("\nPubsub Message: %v\n", string(b))
+	}
+	c += fmt.Sprintf("\nRaw Message: %+v\n", msg)
+	log.Printf("Mail Content:\n %v\n", c)
+	return c
+}