Merge pull request #28 from redBorder/release/1.2.0

Release/1.2.0

Author: Bigomby

Makefile

@@ -37,6 +37,8 @@ coverage:
 	@printf "$(MKL_BLUE)[COVERAGE]$(MKL_CLR_RESET) Computing coverage\n"
 	@echo "mode: count" > coverage.out
 	@go test -covermode=count -coverprofile=counter.part ./counter
+	@go test -covermode=count -coverprofile=monitor.part ./monitor
+	@go test -covermode=count -coverprofile=producer.part ./producer
 	@grep -h -v "mode: count" *.part >> coverage.out
 	@go tool cover -func coverage.out
 

cmd/counters_monitor.go

@@ -19,6 +19,9 @@
 package main
 
 import (
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	rdkafka "github.com/confluentinc/confluent-kafka-go/kafka"
@@ -33,6 +36,12 @@ var reset = make(chan struct{})
 func CountersMonitor(config *AppConfig) {
 	log := log.WithField("prefix", "monitor")
 
+	reload := make(chan os.Signal, 1)
+	signal.Notify(reload, syscall.SIGHUP)
+
+	usr := make(chan os.Signal, 1)
+	signal.Notify(usr, syscall.SIGUSR1)
+
 	wg.Add(1)
 	go func() {
 		for {
@@ -48,29 +57,59 @@ func CountersMonitor(config *AppConfig) {
 				log.Fatalln("Error loading licenses: " + err.Error())
 			}
 
+			for k, v := range limitBytes.getOrganizationLimits() {
+				log.
+					WithField("Total bytes", v).
+					Infof("Organization %s", k)
+			}
+
 			pipeline := BootstrapMonitorPipeline(config, limitBytes)
 			StartConsumingMonitor(pipeline, config)
 
 			log.
 				WithField("Time", intervalEnd.String()).
 				Infof("Next reset set")
 
-			<-time.After(remaining)
+		loop:
+			for {
+				select {
+				case <-usr:
+					pipeline.Produce(nil, map[string]interface{}{"show_total": true}, nil)
+					continue loop
 
-			for org, bytes := range limitBytes {
-				if bytes > 0 {
-					pipeline.Produce(nil, map[string]interface{}{
-						"reset_notification": true,
-						"organization_uuid":  org,
-					}, nil)
+				case <-time.After(remaining):
+					notify(pipeline, limitBytes, true)
+					break loop
+
+				case <-reload:
+					notify(pipeline, limitBytes, false)
+					break loop
 				}
 			}
-
-			reset <- struct{}{}
 		}
 	}()
 }
 
+func notify(
+	pipeline *rbforwarder.RBForwarder, limitBytes LimitBytes, resetCounters bool,
+) {
+	var uuids []string
+
+	for uuid, license := range limitBytes {
+		if !license.Expired {
+			uuids = append(uuids, uuid)
+		}
+	}
+
+	pipeline.Produce(nil, map[string]interface{}{
+		"allowed_licenses": true,
+		"licenses":         uuids,
+		"reset_counters":   resetCounters,
+	}, nil)
+
+	reset <- struct{}{}
+}
+
 // BootstrapMonitorPipeline bootstrap a RBForwarder pipeline
 func BootstrapMonitorPipeline(config *AppConfig, limits LimitBytes) *rbforwarder.RBForwarder {
 	p, err := BootstrapRdKafkaProducer(config.Monitor.Kafka.Attributes)
@@ -90,7 +129,7 @@ func BootstrapMonitorPipeline(config *AppConfig, limits LimitBytes) *rbforwarder
 	components = append(components, &monitor.CountersMonitor{
 		Config: monitor.Config{
 			Workers: 1,
-			Limits:  limits,
+			Limits:  limits.getOrganizationLimits(),
 			Period:  config.Monitor.Timer.Period,
 			Offset:  config.Monitor.Timer.Offset,
 			Log:     log,

cmd/kafka.go

@@ -25,7 +25,9 @@ func BootstrapRdKafkaConsumer(
 	attributes map[string]string, topicAttributes map[string]string,
 ) (*rdkafka.Consumer, error) {
 
-	ta := rdkafka.ConfigMap{}
+	ta := rdkafka.ConfigMap{
+		"auto.offset.reset": "smallest",
+	}
 	for key, value := range topicAttributes {
 		ta.SetKey(key, value)
 	}

cmd/license.go

@@ -38,8 +38,26 @@ import (
 // PubKey should be set on compile time
 var PubKey string
 
+type licenseStatus struct {
+	OrganizationUUID string
+	Amount           uint64
+	Expired          bool
+}
+
 // LimitBytes is a map containing the max bytes for each organization
-type LimitBytes map[string]uint64
+type LimitBytes map[string]*licenseStatus
+
+func (l LimitBytes) getOrganizationLimits() map[string]uint64 {
+	total := make(map[string]uint64)
+
+	for _, v := range l {
+		if !v.Expired {
+			total[v.OrganizationUUID] += v.Amount
+		}
+	}
+
+	return total
+}
 
 // License contains information about the limits of bytes that an organization
 // can send.
@@ -55,7 +73,8 @@ type License struct {
 // LoadLicenses reads a redBorder license from a file and returns a License
 // struct holding the decoded information.
 func LoadLicenses(config *AppConfig) (LimitBytes, error) {
-	var licenses []License
+	limits := make(LimitBytes)
+
 	log := log.WithField("prefix", "license")
 
 	files, err := ioutil.ReadDir(config.LicensesDirectory)
@@ -93,47 +112,36 @@ func LoadLicenses(config *AppConfig) (LimitBytes, error) {
 			return nil, err
 		}
 
+		limits[license.UUID] = &licenseStatus{}
+
 		expires := time.Unix(license.ExpireAt, 0)
 		if expires.Before(time.Now()) {
 			log.Warnf("License %s has expired", license.UUID)
+			limits[license.UUID].Expired = true
 			continue
 		}
 
 		log.Infoln(FormatLicense(license))
 
-		licenses = append(licenses, *license)
-	}
-
-	limits := make(LimitBytes)
-	if !config.OrganizationMode {
-		var totalBytes uint64
-		for _, license := range licenses {
-			if license.Organization != "" {
-				log.Warnf("Ignoring license WITH organization %s", license.UUID)
+		if config.OrganizationMode {
+			if license.Organization == "" {
+				log.Warnf("Ignoring license WITH NO organization %s", license.UUID)
 				continue
 			}
 
-			totalBytes += license.LimitBytes
-		}
-
-		limits["*"] = totalBytes
-	} else {
-		for _, license := range licenses {
-			if license.Organization == "" {
-				log.Warnf("Ignoring license WITH NO organization %s", license.UUID)
+			limits[license.UUID].Amount += license.LimitBytes
+			limits[license.UUID].OrganizationUUID = license.Organization
+		} else {
+			if license.Organization != "" {
+				log.Warnf("Ignoring license WITH organization %s", license.UUID)
 				continue
 			}
 
-			limits[license.Organization] = limits[license.Organization] + license.LimitBytes
+			limits[license.UUID].Amount += license.LimitBytes
+			limits[license.UUID].OrganizationUUID = "*"
 		}
 	}
 
-	for k, v := range limits {
-		log.
-			WithField("Total bytes", v).
-			Infof("Organization %s", k)
-	}
-
 	return limits, nil
 }
 
@@ -237,12 +245,14 @@ func FormatLicense(license *License) string {
 	keyColor := color.New(color.FgYellow, color.Bold).SprintFunc()
 
 	return fmt.Sprintf("Using license \n"+
-		keyColor("\tUUID:         ")+"%s\n"+
-		keyColor("\tCluster UUID: ")+"%s\n"+
-		keyColor("\tExpires:      ")+"%s\n"+
-		keyColor("\tLimit bytes:  ")+"%d\n",
+		keyColor("\tUUID:              ")+"%s\n"+
+		keyColor("\tCluster UUID:      ")+"%s\n"+
+		keyColor("\tOrganization UUID: ")+"%s\n"+
+		keyColor("\tExpires:           ")+"%s\n"+
+		keyColor("\tLimit bytes:       ")+"%d",
 		license.UUID,
 		license.ClusterUUID,
+		license.Organization,
 		time.Unix(license.ExpireAt, 0).String(),
 		license.LimitBytes,
 	)

monitor/messages.go

@@ -30,10 +30,11 @@ type Count struct {
 // Alert contains the information abot a message alerting that the
 // maximum number of messages has been reached.
 type Alert struct {
-	Monitor      string `json:"monitor"`
-	Type         string `json:"type"`
-	UUID         string `json:"uuid,omitempty"`
-	CurrentBytes uint64 `json:"current_bytes,omitempty"`
-	Limit        uint64 `json:"limit"`
-	Timestamp    int64  `json:"timestamp"`
+	Monitor      string   `json:"monitor"`
+	Type         string   `json:"type"`
+	UUID         string   `json:"uuid,omitempty"`
+	CurrentBytes uint64   `json:"current_bytes,omitempty"`
+	Limit        uint64   `json:"limit,omitempty"`
+	Timestamp    int64    `json:"timestamp"`
+	Licenses     []string `json:"licenses,omitempty"`
 }

monitor/monitor.go

@@ -25,11 +25,15 @@ import (
 
 type logger interface {
 	Debugf(format string, args ...interface{})
+	Infof(format string, args ...interface{})
+	Infoln(args ...interface{})
 }
 
 type nullLogger struct{}
 
 func (n *nullLogger) Debugf(format string, args ...interface{}) {}
+func (n *nullLogger) Infof(format string, args ...interface{})  {}
+func (n *nullLogger) Infoln(args ...interface{})                {}
 
 // Config contains the configuration for a Monitor.
 type Config struct {
@@ -64,6 +68,7 @@ func (mon *CountersMonitor) Spawn(id int) utils.Composer {
 		monitor.Log = new(nullLogger)
 	}
 	monitor.db = bootstrapDB(mon.Limits)
+
 	return monitor
 }
 
@@ -76,19 +81,40 @@ func (mon *CountersMonitor) Spawn(id int) utils.Composer {
 //   does, send an alert to Kafka.
 func (mon *CountersMonitor) OnMessage(m *utils.Message, done utils.Done) {
 	var (
-		ok      bool
 		payload []byte
 		err     error
 		bytes   uint64
 	)
 
-	if _, ok = m.Opts.Get("reset_notification"); ok {
-		org, _ := m.Opts.Get("organization_uuid")
-		mon.db[org.(string)] = 0
+	if showTotal, ok := m.Opts.Get("show_total"); ok {
+		if showTotalBool, ok := showTotal.(bool); ok {
+			if showTotalBool {
+				for k, v := range mon.db {
+					mon.Log.Infof("[%s] Consumed bytes %d", k, v)
+				}
+			}
+		}
+
+		done(m, 0, "Show total")
+		return
+	}
 
-		m.PushPayload(createResetNotificationMessage(org.(string)))
-		mon.Log.Debugf("Sending reset notification")
-		done(m, 0, "Reset notification")
+	if _, ok := m.Opts.Get("allowed_licenses"); ok {
+		if resetCounters, ok := m.Opts.Get("reset_counters"); ok {
+			if shouldReset, ok := resetCounters.(bool); ok {
+				if shouldReset {
+					for organization := range mon.db {
+						mon.db[organization] = 0
+					}
+					mon.Log.Infoln("Counters has been reset")
+				}
+			}
+		}
+
+		licenses, _ := m.Opts.Get("licenses")
+		// FIXME check assertion
+		m.PushPayload(createLicensesAllowedMessage(licenses.([]string)))
+		done(m, 0, "Allowed licenses")
 		return
 	}
 
@@ -103,11 +129,12 @@ func (mon *CountersMonitor) OnMessage(m *utils.Message, done utils.Done) {
 		return
 	}
 
-	if ok = belongsToInterval(count.Timestamp, mon.Period, mon.Offset, mon.clk.Now().Unix()); !ok {
+	if ok := belongsToInterval(count.Timestamp, mon.Period, mon.Offset, mon.clk.Now().Unix()); !ok {
 		done(m, 0, "Message too old")
 		return
 	}
 
+	var ok bool
 	if bytes, ok = mon.db[count.UUID]; !ok {
 		m.PushPayload(createUknownUUIDMessage(count.UUID))
 		done(m, 0, "Unknown UUID: \""+count.UUID+"\"")