Merge pull request #12 from cosmo0920/use-logrus-logger

Use logrus logger instead of fmt

Author: cosmo0920

Gopkg.lock

@@ -7,13 +7,13 @@ import "github.com/aws/aws-sdk-go/aws/awserr"
 import "github.com/aws/aws-sdk-go/aws/session"
 import "github.com/aws/aws-sdk-go/service/s3"
 import "github.com/aws/aws-sdk-go/service/s3/s3manager"
+import log "github.com/sirupsen/logrus"
 import "github.com/prometheus/common/version"
 
 import (
 	"C"
 	"bytes"
 	"compress/gzip"
-	"fmt"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -23,12 +23,19 @@ import (
 )
 
 var plugin GoOutputPlugin = &fluentPlugin{}
+var logger *log.Logger
+
+func init() {
+	logLevel, _ := log.ParseLevel("info")
+	logger = newLogger(logLevel)
+}
 
 type s3operator struct {
 	bucket         string
 	prefix         string
 	uploader       *s3manager.Uploader
 	compressFormat format
+	logger         *log.Logger
 }
 
 type GoOutputPlugin interface {
@@ -130,15 +137,19 @@ func ensureBucket(session *session.Session, bucket, region *string) (bool, error
 		}
 	}
 
-	_, err := svc.CreateBucket(input)
+	result, err := svc.CreateBucket(input)
+	logger.Tracef("CreateBucket request result is: %s, err: %s", result, err)
 	if err != nil {
 		if aerr, ok := err.(awserr.Error); ok {
 			switch aerr.Code() {
 			case s3.ErrCodeBucketAlreadyExists:
+				logger.Tracef("Bucket(%s) is already exists.", *bucket)
 				return true, nil
 			case s3.ErrCodeBucketAlreadyOwnedByYou:
+				logger.Tracef("Bucket(%s) is already owned by you.", *bucket)
 				return true, nil
 			default:
+				logger.Tracef("CreateBucket is failed with: %s", aerr.Error())
 				return false, aerr
 			}
 		} else {
@@ -148,6 +159,12 @@ func ensureBucket(session *session.Session, bucket, region *string) (bool, error
 	return true, nil
 }
 
+func newLogger(logLevel log.Level) *log.Logger {
+	logger := log.New()
+	logger.Level = logLevel
+	return logger
+}
+
 func newS3Output(ctx unsafe.Pointer, operatorID int) (*s3operator, error) {
 	// Example to retrieve an optional configuration parameter
 	credential := plugin.PluginConfigKey(ctx, "Credential")
@@ -159,21 +176,24 @@ func newS3Output(ctx unsafe.Pointer, operatorID int) (*s3operator, error) {
 	compress := plugin.PluginConfigKey(ctx, "Compress")
 	endpoint := plugin.PluginConfigKey(ctx, "Endpoint")
 	autoCreateBucket := plugin.PluginConfigKey(ctx, "AutoCreateBucket")
+	logLevel := plugin.PluginConfigKey(ctx, "LogLevel")
 
-	config, err := getS3Config(accessKeyID, secretAccessKey, credential, s3prefix, bucket, region, compress, endpoint, autoCreateBucket)
+	config, err := getS3Config(accessKeyID, secretAccessKey, credential, s3prefix, bucket, region, compress, endpoint, autoCreateBucket, logLevel)
 	if err != nil {
 		return nil, err
 	}
-	fmt.Printf("[flb-go %d] Starting fluent-bit-go-s3: %s\n", operatorID, version.Info())
-	fmt.Printf("[flb-go %d] plugin credential parameter = '%s'\n", operatorID, credential)
-	fmt.Printf("[flb-go %d] plugin accessKeyID parameter = '%s'\n", operatorID, accessKeyID)
-	fmt.Printf("[flb-go %d] plugin secretAccessKey parameter = '%s'\n", operatorID, secretAccessKey)
-	fmt.Printf("[flb-go %d] plugin bucket parameter = '%s'\n", operatorID, bucket)
-	fmt.Printf("[flb-go %d] plugin s3prefix parameter = '%s'\n", operatorID, s3prefix)
-	fmt.Printf("[flb-go %d] plugin region parameter = '%s'\n", operatorID, region)
-	fmt.Printf("[flb-go %d] plugin compress parameter = '%s'\n", operatorID, compress)
-	fmt.Printf("[flb-go %d] plugin endpoint parameter = '%s'\n", operatorID, endpoint)
-	fmt.Printf("[flb-go %d] plugin autoCreateBucket parameter = '%s'\n", operatorID, autoCreateBucket)
+	logger := newLogger(config.logLevel)
+
+	logger.Infof("[flb-go %d] Starting fluent-bit-go-s3: %v\n", operatorID, version.Info())
+	logger.Infof("[flb-go %d] plugin credential parameter = '%s'\n", operatorID, credential)
+	logger.Infof("[flb-go %d] plugin accessKeyID parameter = '%s'\n", operatorID, accessKeyID)
+	logger.Infof("[flb-go %d] plugin secretAccessKey parameter = '%s'\n", operatorID, secretAccessKey)
+	logger.Infof("[flb-go %d] plugin bucket parameter = '%s'\n", operatorID, bucket)
+	logger.Infof("[flb-go %d] plugin s3prefix parameter = '%s'\n", operatorID, s3prefix)
+	logger.Infof("[flb-go %d] plugin region parameter = '%s'\n", operatorID, region)
+	logger.Infof("[flb-go %d] plugin compress parameter = '%s'\n", operatorID, compress)
+	logger.Infof("[flb-go %d] plugin endpoint parameter = '%s'\n", operatorID, endpoint)
+	logger.Infof("[flb-go %d] plugin autoCreateBucket parameter = '%s'\n", operatorID, autoCreateBucket)
 
 	cfg := aws.Config{
 		Credentials: config.credentials,
@@ -202,6 +222,7 @@ func newS3Output(ctx unsafe.Pointer, operatorID int) (*s3operator, error) {
 		prefix:         *config.s3prefix,
 		uploader:       uploader,
 		compressFormat: config.compress,
+		logger:         logger,
 	}
 
 	return s3operator, nil
@@ -210,7 +231,7 @@ func newS3Output(ctx unsafe.Pointer, operatorID int) (*s3operator, error) {
 
 func addS3Output(ctx unsafe.Pointer) error {
 	operatorID := len(s3operators)
-	fmt.Printf("[s3operator] id = %q\n", operatorID)
+	logger.Infof("[s3operator] id = %d\n", operatorID)
 	// Set the context to point to any Go variable
 	output.FLBPluginSetContext(ctx, operatorID)
 	operator, err := newS3Output(ctx, operatorID)
@@ -263,7 +284,7 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
 
 		line, err := createJSON(record)
 		if err != nil {
-			fmt.Printf("error creating message for S3: %v\n", err)
+			s3operator.logger.Warnf("error creating message for S3: %v\n", err)
 			continue
 		}
 		lines += line + "\n"
@@ -272,7 +293,7 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
 	objectKey := GenerateObjectKey(s3operator, time.Now())
 	err := plugin.Put(s3operator, objectKey, time.Now(), lines)
 	if err != nil {
-		fmt.Printf("error sending message for S3: %v\n", err)
+		s3operator.logger.Warnf("error sending message for S3: %v\n", err)
 		return output.FLB_RETRY
 	}
 

out_s3.go

@@ -162,6 +162,7 @@ type testFluentPlugin struct {
 	compress         string
 	endpoint         string
 	autoCreateBucket string
+	logLevel         string
 	records          []testrecord
 	position         int
 	events           []*events
@@ -187,6 +188,8 @@ func (p *testFluentPlugin) PluginConfigKey(ctx unsafe.Pointer, key string) strin
 		return p.endpoint
 	case "AutoCreateBucket":
 		return p.autoCreateBucket
+	case "LogLevel":
+		return p.logLevel
 	}
 	return "unknown-" + key
 }
@@ -246,7 +249,7 @@ func (c *testS3Credential) GetCredentials(accessID, secretkey, credential string
 
 func TestPluginInitializationWithStaticCredentials(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	_, err := getS3Config("exampleaccessID", "examplesecretkey", "", "exampleprefix", "examplebucket", "exampleregion", "", "", "false")
+	_, err := getS3Config("exampleaccessID", "examplesecretkey", "", "exampleprefix", "examplebucket", "exampleregion", "", "", "false", "info")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -259,14 +262,15 @@ func TestPluginInitializationWithStaticCredentials(t *testing.T) {
 		compress:         "",
 		endpoint:         "",
 		autoCreateBucket: "false",
+		logLevel:         "info",
 	}
 	res := FLBPluginInit(unsafe.Pointer(&plugin))
 	assert.Equal(t, output.FLB_OK, res)
 }
 
 func TestPluginInitializationWithSharedCredentials(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	_, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "", "", "false")
+	_, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "", "", "false", "info")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -278,6 +282,7 @@ func TestPluginInitializationWithSharedCredentials(t *testing.T) {
 		compress:         "",
 		endpoint:         "",
 		autoCreateBucket: "false",
+		logLevel:         "info",
 	}
 	res := FLBPluginInit(unsafe.Pointer(&plugin))
 	assert.Equal(t, output.FLB_OK, res)

out_s3_test.go

@@ -2,6 +2,7 @@ package main
 
 import "github.com/aws/aws-sdk-go/aws"
 import "github.com/aws/aws-sdk-go/aws/credentials"
+import log "github.com/sirupsen/logrus"
 
 import (
 	"fmt"
@@ -23,6 +24,7 @@ type s3Config struct {
 	region           *string
 	compress         format
 	endpoint         string
+	logLevel         log.Level
 	autoCreateBucket bool
 }
 
@@ -62,7 +64,7 @@ func (c *s3PluginConfig) GetCredentials(accessKeyID, secretKey, credential strin
 	return nil, fmt.Errorf("Failed to create credentials")
 }
 
-func getS3Config(accessID, secretKey, credential, s3prefix, bucket, region, compress, endpoint, autoCreateBucket string) (*s3Config, error) {
+func getS3Config(accessID, secretKey, credential, s3prefix, bucket, region, compress, endpoint, autoCreateBucket, logLevel string) (*s3Config, error) {
 	conf := &s3Config{}
 	creds, err := s3Creds.GetCredentials(accessID, secretKey, credential)
 	if err != nil {
@@ -106,5 +108,14 @@ func getS3Config(accessID, secretKey, credential, s3prefix, bucket, region, comp
 		conf.autoCreateBucket = isAutoCreateBucket
 	}
 
+	if logLevel == "" {
+		logLevel = "info"
+	}
+	var level log.Level
+	if level, err = log.ParseLevel(logLevel); err != nil {
+		return nil, fmt.Errorf("invalid log level: %v", logLevel)
+	}
+	conf.logLevel = level
+
 	return conf, nil
 }

s3.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestGetS3ConfigStaticCredentials(t *testing.T) {
-	conf, err := getS3Config("exampleaccessID", "examplesecretkey", "", "exampleprefix", "examplebucket", "exampleregion", "", "", "")
+	conf, err := getS3Config("exampleaccessID", "examplesecretkey", "", "exampleprefix", "examplebucket", "exampleregion", "", "", "", "")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -23,7 +23,7 @@ func TestGetS3ConfigStaticCredentials(t *testing.T) {
 
 func TestGetS3ConfigSharedCredentials(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "", "", "")
+	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "", "", "", "")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -38,7 +38,7 @@ func TestGetS3ConfigSharedCredentials(t *testing.T) {
 
 func TestGetS3ConfigCompression(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "", "")
+	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "", "", "")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -53,7 +53,7 @@ func TestGetS3ConfigCompression(t *testing.T) {
 
 func TestGetS3ConfigEndpoint(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "http://localhost:9000", "false")
+	conf, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "http://localhost:9000", "false", "")
 	if err != nil {
 		t.Fatalf("failed test %#v", err)
 	}
@@ -69,7 +69,7 @@ func TestGetS3ConfigEndpoint(t *testing.T) {
 
 func TestGetS3ConfigInvalidEndpoint(t *testing.T) {
 	s3Creds = &testS3Credential{}
-	_, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "https://your-bucketname.s3.amazonaws.com", "false")
+	_, err := getS3Config("", "", "examplecredentials", "exampleprefix", "examplebucket", "exampleregion", "gzip", "https://your-bucketname.s3.amazonaws.com", "false", "")
 	if err != nil {
 		expected := errors.New("Endpoint is not supported for AWS S3. This parameter is intended for S3 compatible services. Use Region instead.")
 		assert.Equal(t, expected, err)