Skip to content

Use interface to log in Consul package #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 14, 2020
Merged

Use interface to log in Consul package #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions consul/logger.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
package consul

// Logger Allows replacing easily the logger.
type Logger interface {
// Debugf Display debug message
Debugf(format string, args ...interface{})
// Infof Display info message
Infof(format string, args ...interface{})
// Warnf Display warning message
Warnf(format string, args ...interface{})
// Errorf Display error message
Errorf(format string, args ...interface{})
}
32 changes: 32 additions & 0 deletions consul/logger_testing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
package consul

import "testing"

type testingLogger struct {
t *testing.T
}

// Debugf Display debug message
func (l *testingLogger) Debugf(format string, args ...interface{}) {
l.t.Logf(format, args...)
}

// Infof Display info message
func (l *testingLogger) Infof(format string, args ...interface{}) {
l.t.Logf(format, args...)
}

// Warnf Display warning message
func (l *testingLogger) Warnf(format string, args ...interface{}) {
l.t.Logf(format, args...)
}

// Errorf Display error message
func (l *testingLogger) Errorf(format string, args ...interface{}) {
l.t.Logf(format, args...)
}

// NewTestingLogger creates a Logger for testing.T
func NewTestingLogger(t *testing.T) Logger {
return &testingLogger{t: t}
}
40 changes: 21 additions & 19 deletions consul/watcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ import (

"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/command/connect/proxy"
log "github.com/sirupsen/logrus"
)

const (
Expand Down Expand Up @@ -58,16 +57,19 @@ type Watcher struct {
leaf *certLeaf

update chan struct{}
log Logger
}

func New(service string, consul *api.Client) *Watcher {
// New builds a new watcher
func New(service string, consul *api.Client, log Logger) *Watcher {
return &Watcher{
service: service,
consul: consul,

C: make(chan Config),
upstreams: make(map[string]*upstream),
update: make(chan struct{}, 1),
log: log,
}
}

Expand Down Expand Up @@ -144,7 +146,7 @@ func (w *Watcher) handleProxyChange(first bool, srv *api.AgentService) {
}

func (w *Watcher) startUpstream(up api.Upstream) {
log.Infof("consul: watching upstream for service %s", up.DestinationName)
w.log.Infof("consul: watching upstream for service %s", up.DestinationName)

u := &upstream{
LocalBindAddress: up.LocalBindAddress,
Expand All @@ -169,7 +171,7 @@ func (w *Watcher) startUpstream(up api.Upstream) {
WaitIndex: index,
})
if err != nil {
log.Errorf("consul: error fetching service definition for service %s: %s", up.DestinationName, err)
w.log.Errorf("consul: error fetching service definition for service %s: %s", up.DestinationName, err)
time.Sleep(errorWaitTime)
index = 0
continue
Expand All @@ -188,7 +190,7 @@ func (w *Watcher) startUpstream(up api.Upstream) {
}

func (w *Watcher) removeUpstream(name string) {
log.Infof("consul: removing upstream for service %s", name)
w.log.Infof("consul: removing upstream for service %s", name)

w.lock.Lock()
w.upstreams[name].done = true
Expand All @@ -197,7 +199,7 @@ func (w *Watcher) removeUpstream(name string) {
}

func (w *Watcher) watchLeaf() {
log.Debugf("consul: watching leaf cert for %s", w.serviceName)
w.log.Debugf("consul: watching leaf cert for %s", w.serviceName)

var lastIndex uint64
first := true
Expand All @@ -207,7 +209,7 @@ func (w *Watcher) watchLeaf() {
WaitIndex: lastIndex,
})
if err != nil {
log.Errorf("consul error fetching leaf cert for service %s: %s", w.serviceName, err)
w.log.Errorf("consul error fetching leaf cert for service %s: %s", w.serviceName, err)
time.Sleep(errorWaitTime)
lastIndex = 0
continue
Expand All @@ -217,7 +219,7 @@ func (w *Watcher) watchLeaf() {
lastIndex = meta.LastIndex

if changed {
log.Infof("consul: leaf cert for service %s changed, serial: %s, valid before: %s, valid after: %s", w.serviceName, cert.SerialNumber, cert.ValidBefore, cert.ValidAfter)
w.log.Infof("consul: leaf cert for service %s changed, serial: %s, valid before: %s, valid after: %s", w.serviceName, cert.SerialNumber, cert.ValidBefore, cert.ValidAfter)
w.lock.Lock()
if w.leaf == nil {
w.leaf = &certLeaf{}
Expand All @@ -229,15 +231,15 @@ func (w *Watcher) watchLeaf() {
}

if first {
log.Infof("consul: leaf cert for %s ready", w.serviceName)
w.log.Infof("consul: leaf cert for %s ready", w.serviceName)
w.ready.Done()
first = false
}
}
}

func (w *Watcher) watchService(service string, handler func(first bool, srv *api.AgentService)) {
log.Infof("consul: watching service %s", service)
w.log.Infof("consul: watching service %s", service)

hash := ""
first := true
Expand All @@ -247,7 +249,7 @@ func (w *Watcher) watchService(service string, handler func(first bool, srv *api
WaitTime: 10 * time.Minute,
})
if err != nil {
log.Errorf("consul: error fetching service %s definition: %s", service, err)
w.log.Errorf("consul: error fetching service %s definition: %s", service, err)
time.Sleep(errorWaitTime)
hash = ""
continue
Expand All @@ -257,7 +259,7 @@ func (w *Watcher) watchService(service string, handler func(first bool, srv *api
hash = meta.LastContentHash

if changed {
log.Debugf("consul: service %s changed", service)
w.log.Debugf("consul: service %s changed", service)
handler(first, srv)
w.notifyChanged()
}
Expand All @@ -267,7 +269,7 @@ func (w *Watcher) watchService(service string, handler func(first bool, srv *api
}

func (w *Watcher) watchCA() {
log.Debugf("consul: watching ca certs")
w.log.Debugf("consul: watching ca certs")

first := true
var lastIndex uint64
Expand All @@ -277,7 +279,7 @@ func (w *Watcher) watchCA() {
WaitTime: 10 * time.Minute,
})
if err != nil {
log.Errorf("consul: error fetching cas: %s", err)
w.log.Errorf("consul: error fetching cas: %s", err)
time.Sleep(errorWaitTime)
lastIndex = 0
continue
Expand All @@ -287,37 +289,37 @@ func (w *Watcher) watchCA() {
lastIndex = meta.LastIndex

if changed {
log.Infof("consul: CA certs changed, active root id: %s", caList.ActiveRootID)
w.log.Infof("consul: CA certs changed, active root id: %s", caList.ActiveRootID)
w.lock.Lock()
w.certCAs = w.certCAs[:0]
w.certCAPool = x509.NewCertPool()
for _, ca := range caList.Roots {
w.certCAs = append(w.certCAs, []byte(ca.RootCertPEM))
ok := w.certCAPool.AppendCertsFromPEM([]byte(ca.RootCertPEM))
if !ok {
log.Warn("consul: unable to add CA certificate to pool")
w.log.Warnf("consul: unable to add CA certificate to pool for root id: %s", caList.ActiveRootID)
}
}
w.lock.Unlock()
w.notifyChanged()
}

if first {
log.Infof("consul: CA certs ready")
w.log.Infof("consul: CA certs ready")
w.ready.Done()
first = false
}
}
}

func (w *Watcher) genCfg() Config {
log.Debug("generating configuration...")
w.log.Debugf("generating configuration for service %s[%s]...", w.serviceName, w.service)
w.lock.Lock()
serviceInstancesAlive := 0
serviceInstancesTotal := 0
defer func() {
w.lock.Unlock()
log.Debugf("done generating configuration, instances: %d/%d total",
w.log.Debugf("done generating configuration, instances: %d/%d total",
serviceInstancesAlive, serviceInstancesTotal)
}()

Expand Down
1 change: 1 addition & 0 deletions haproxy/haproxy_cmd/run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ func Start(sd *lib.Shutdown, cfg Config) (*dataplane.Dataplane, error) {

err = dataplaneClient.Ping()
if err != nil {
fmt.Println("*****\n* SOUCHAY: wait for dataplane to be up\n*****")
time.Sleep(100 * time.Millisecond)
continue
}
Expand Down
25 changes: 24 additions & 1 deletion main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,28 @@ import (
"github.com/criteo/haproxy-consul-connect/consul"
)

type consulLogger struct{}

// Debugf Display debug message
func (consulLogger) Debugf(format string, args ...interface{}) {
log.Debugf(format, args...)
}

// Infof Display info message
func (consulLogger) Infof(format string, args ...interface{}) {
log.Infof(format, args...)
}

// Warnf Display warning message
func (consulLogger) Warnf(format string, args ...interface{}) {
log.Infof(format, args...)
}

// Errorf Display error message
func (consulLogger) Errorf(format string, args ...interface{}) {
log.Errorf(format, args...)
}

func main() {
logLevel := flag.String("log-level", "INFO", "Log level")
consulAddr := flag.String("http-addr", "127.0.0.1:8500", "Consul agent address")
Expand Down Expand Up @@ -73,7 +95,8 @@ func main() {
log.Fatalf("Please specify -sidecar-for or -sidecar-for-tag")
}

watcher := consul.New(serviceID, consulClient)
consulLogger := &consulLogger{}
watcher := consul.New(serviceID, consulClient, consulLogger)
go func() {
if err := watcher.Run(); err != nil {
log.Error(err)
Expand Down
2 changes: 1 addition & 1 deletion utils_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ func startConnectService(t *testing.T, sd *lib.Shutdown, client *api.Client, reg

errs := make(chan error, 2)

watcher := consul.New(reg.ID, client)
watcher := consul.New(reg.ID, client, consul.NewTestingLogger(t))
go func() {
err := watcher.Run()
if err != nil {
Expand Down