// Copyright 2015 clair 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 notifier fetches notifications from the database and sends them // to the specified remote handler. package notifier import ( "time" "github.com/coreos/pkg/capnslog" "github.com/coreos/pkg/timeutil" "github.com/pborman/uuid" "github.com/coreos/clair/config" "github.com/coreos/clair/database" "github.com/coreos/clair/health" "github.com/coreos/clair/utils" ) var log = capnslog.NewPackageLogger("github.com/coreos/clair", "notifier") const ( checkInterval = 5 * time.Minute refreshLockDuration = time.Minute * 2 lockDuration = time.Minute*8 + refreshLockDuration maxBackOff = 15 * time.Minute ) // TODO(Quentin-M): Allow registering custom notification handlers. // A Notification represents the structure of the notifications that are sent by a Notifier. type Notification struct { Name, Type string Content interface{} } var notifiers = make(map[string]Notifier) // Notifier represents anything that can transmit notifications. type Notifier interface { // Configure attempts to initialize the notifier with the provided configuration. // It returns whether the notifier is enabled or not. Configure(*config.NotifierConfig) (bool, error) // Send transmits the specified notification. Send(notification *Notification) error } // RegisterNotifier makes a Fetcher available by the provided name. // If Register is called twice with the same name or if driver is nil, // it panics. func RegisterNotifier(name string, n Notifier) { if name == "" { panic("notifier: could not register a Notifier with an empty name") } if n == nil { panic("notifier: could not register a nil Notifier") } if _, dup := notifiers[name]; dup { panic("notifier: RegisterNotifier called twice for " + name) } notifiers[name] = n } // Run starts the Notifier service. func Run(config *config.NotifierConfig, stopper *utils.Stopper) { defer stopper.End() // Configure registered notifiers. for notifierName, notifier := range notifiers { if configured, err := notifier.Configure(config); configured { log.Infof("notifier '%s' configured\n", notifierName) } else { delete(notifiers, notifierName) if err != nil { log.Errorf("could not configure notifier '%s': %s", notifierName, err) } } } // Do not run the updater if there is no notifier enabled. if len(notifiers) == 0 { log.Infof("notifier service is disabled") return } whoAmI := uuid.New() log.Infof("notifier service started. lock identifier: %s\n", whoAmI) // Register healthchecker. health.RegisterHealthchecker("notifier", Healthcheck) for { // Find task. // TODO(Quentin-M): Combine node and notification. node, notification := findTask(whoAmI, stopper) if node == "" && notification == nil { break } // Handle task. done := make(chan bool, 1) go func() { if handleTask(notification, stopper, config.Attempts) { database.MarkNotificationAsSent(node) } database.Unlock(node, whoAmI) done <- true }() // Refresh task lock until done. outer: for { select { case <-done: break outer case <-time.After(refreshLockDuration): database.Lock(node, lockDuration, whoAmI) } } } log.Info("notifier service stopped") } func findTask(whoAmI string, stopper *utils.Stopper) (string, database.Notification) { for { // Find a notification to send. node, notification, err := database.FindOneNotificationToSend(database.GetDefaultNotificationWrapper()) if err != nil { log.Warningf("could not get notification to send: %s", err) } // No notification or error: wait. if notification == nil || err != nil { if !stopper.Sleep(checkInterval) { return "", nil } continue } // Lock the notification. if hasLock, _ := database.Lock(node, lockDuration, whoAmI); hasLock { log.Infof("found and locked a notification: %s", notification.GetName()) return node, notification } } } func handleTask(notification database.Notification, st *utils.Stopper, maxAttempts int) bool { // Get notification content. // TODO(Quentin-M): Split big notifications. notificationContent, err := notification.GetContent() if err != nil { log.Warningf("could not get content of notification '%s': %s", notification.GetName(), err) return false } // Create notification. payload := &Notification{ Name: notification.GetName(), Type: notification.GetType(), Content: notificationContent, } // Send notification. for notifierName, notifier := range notifiers { var attempts int var backOff time.Duration for { // Max attempts exceeded. if attempts >= maxAttempts { log.Infof("giving up on sending notification '%s' to notifier '%s': max attempts exceeded (%d)\n", notification.GetName(), notifierName, maxAttempts) return false } // Backoff. if backOff > 0 { log.Infof("waiting %v before retrying to send notification '%s' to notifier '%s' (Attempt %d / %d)\n", backOff, notification.GetName(), notifierName, attempts+1, maxAttempts) if !st.Sleep(backOff) { return false } } // Send using the current notifier. if err := notifier.Send(payload); err == nil { // Send has been successful. Go to the next one. break } // Send failed; increase attempts/backoff and retry. log.Errorf("could not send notification '%s' to notifier '%s': %s", notification.GetName(), notifierName, err) backOff = timeutil.ExpBackoff(backOff, maxBackOff) attempts++ } } log.Infof("successfully sent notification '%s'\n", notification.GetName()) return true } // Healthcheck returns the health of the notifier service. func Healthcheck() health.Status { queueSize, err := database.CountNotificationsToSend() return health.Status{IsEssential: false, IsHealthy: err == nil, Details: struct{ QueueSize int }{QueueSize: queueSize}} }