Kafka plugin and bugs fixed (#242)

* Turn kafka into a plugin

* Re-modification of CopyObject and bug fixes for ListObject

Author: fireworkmarks

.travis.yml

@@ -37,9 +37,6 @@ script:
 - go test -v
 - popd
 - cat lc.log
-- pushd messagebus/tests
-- go test -check.vv
-- popd
 
 before_deploy:
   - echo "prepare package"

api/access-log-handler.go

@@ -7,9 +7,8 @@ import (
 	"time"
 
 	"github.com/journeymidnight/yig/helper"
-	bus "github.com/journeymidnight/yig/messagebus"
-	"github.com/journeymidnight/yig/messagebus/types"
 	"github.com/journeymidnight/yig/meta"
+	bus "github.com/journeymidnight/yig/mq"
 )
 
 type ResponseRecorder struct {
@@ -56,15 +55,12 @@ func (a AccessLogHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	response := newReplacer.Replace(a.format)
 
 	helper.AccessLogger.Println(response)
-	// send the entries in access logger to message bus.
+	// send the entries in access logger to message queue.
 	elems := newReplacer.GetReplacedValues()
 	a.notify(elems)
 }
 
 func (a AccessLogHandler) notify(elems map[string]string) {
-	if !helper.CONFIG.MsgBus.Enabled {
-		return
-	}
 	if len(elems) == 0 {
 		return
 	}
@@ -74,29 +70,14 @@ func (a AccessLogHandler) notify(elems map[string]string) {
 		return
 	}
 
-	sender, err := bus.GetMessageSender()
-	if err != nil {
-		helper.Logger.Error("Failed to get message bus sender, err:", err)
-		return
-	}
-
-	// send the message to message bus async.
-	// don't set the ErrChan.
-	msg := &types.Message{
-		Topic:   helper.CONFIG.MsgBus.Topic,
-		Key:     "",
-		ErrChan: nil,
-		Value:   val,
-	}
-
-	err = sender.AsyncSend(msg)
+	err = bus.MsgSender.AsyncSend(val)
 	if err != nil {
 		helper.Logger.Error(
-			fmt.Sprintf("Failed to send message [%v] to message bus, err: %v",
+			fmt.Sprintf("Failed to send message [%v] to message queue, err: %v",
 				elems, err))
 		return
 	}
-	helper.Logger.Info(fmt.Sprintf("Succeed to send message [%v] to message bus.",
+	helper.Logger.Info(fmt.Sprintf("Succeed to send message [%v] to message queue.",
 		elems))
 }
 

api/object-handlers.go

@@ -492,25 +492,52 @@ func (api ObjectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		return
 	}
 
+	// TODO: In a versioning-enabled bucket, you cannot change the storage class
+	//  of a specific version of an object. When you copy it, Amazon S3 gives it
+	//  a new version ID.
+	storageClassStr := r.Header.Get("X-Amz-Storage-Class")
+	var targetStorageClass meta.StorageClass
+	if storageClassStr != "" {
+		helper.Logger.Info("Get storage class header:", storageClassStr)
+		targetStorageClass, err = meta.MatchStorageClassIndex(storageClassStr)
+		if err != nil {
+			WriteErrorResponse(w, r, err)
+			return
+		}
+	} else {
+		targetStorageClass = sourceObject.StorageClass
+	}
+	if targetStorageClass == meta.ObjectStorageClassGlacier || targetStorageClass == meta.ObjectStorageClassDeepArchive {
+		WriteErrorResponse(w, r, ErrInvalidCopySourceStorageClass)
+		return
+	}
+
 	// maximum Upload size for object in a single CopyObject operation.
 	if isMaxObjectSize(sourceObject.Size) {
 		WriteErrorResponseWithResource(w, r, ErrEntityTooLarge, copySource)
 		return
 	}
 
+	var isMetadataOnly bool
+	if sourceBucketName == targetBucketName && sourceObjectName == targetObjectName {
+		isMetadataOnly = true
+	}
+
 	pipeReader, pipeWriter := io.Pipe()
-	go func() {
-		startOffset := int64(0) // Read the whole file.
-		// Get the object.
-		err = api.ObjectAPI.GetObject(sourceObject, startOffset, sourceObject.Size,
-			pipeWriter, sseRequest)
-		if err != nil {
-			logger.Error("Unable to read an object:", err)
-			pipeWriter.CloseWithError(err)
-			return
-		}
-		pipeWriter.Close()
-	}()
+	if !isMetadataOnly {
+		go func() {
+			startOffset := int64(0) // Read the whole file.
+			// Get the object.
+			err = api.ObjectAPI.GetObject(sourceObject, startOffset, sourceObject.Size,
+				pipeWriter, sseRequest)
+			if err != nil {
+				logger.Error("Unable to read an object:", err)
+				pipeWriter.CloseWithError(err)
+				return
+			}
+			pipeWriter.Close()
+		}()
+	}
 
 	targetACL, err := getAclFromHeader(r.Header)
 	if err != nil {
@@ -527,55 +554,25 @@ func (api ObjectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	targetObject.Etag = sourceObject.Etag
 	targetObject.Parts = sourceObject.Parts
 	targetObject.Type = sourceObject.Type
+	targetObject.StorageClass = targetStorageClass
 
 	directive := r.Header.Get("X-Amz-Metadata-Directive")
 	if directive == "COPY" || directive == "" {
 		targetObject.CustomAttributes = sourceObject.CustomAttributes
-		targetObject.StorageClass = sourceObject.StorageClass
 		targetObject.ContentType = sourceObject.ContentType
 	} else if directive == "REPLACE" {
-		// TODO: In a versioning-enabled bucket, you cannot change the storage class
-		//  of a specific version of an object. When you copy it, Amazon S3 gives it
-		//  a new version ID.
-		storageClassFromHeader, err := getStorageClassFromHeader(r)
-		if err != nil {
-			WriteErrorResponse(w, r, err)
-			return
-		}
-		if storageClassFromHeader == meta.ObjectStorageClassGlacier || storageClassFromHeader == meta.ObjectStorageClassDeepArchive {
-			WriteErrorResponse(w, r, ErrInvalidCopySourceStorageClass)
-			return
-		}
 		newMetadata := extractMetadataFromHeader(r.Header)
 		if c, ok := newMetadata["content-type"]; ok {
 			targetObject.ContentType = c
 		} else {
 			targetObject.ContentType = sourceObject.ContentType
 		}
 		targetObject.CustomAttributes = newMetadata
-		targetObject.StorageClass = storageClassFromHeader
 	} else {
 		WriteErrorResponse(w, r, ErrInvalidCopyRequest)
 		return
 	}
 
-	var isMetadataOnly bool
-	if sourceBucketName == targetBucketName && sourceObjectName == targetObjectName {
-		isMetadataOnly = true
-	}
-
-	// Check if x-amz-metadata-directive was not set to REPLACE and source,
-	// desination are same objects. Apply this restriction also when
-	// metadataOnly is true indicating that we are not overwriting the object.
-	// if encryption is enabled we do not need explicit "REPLACE" metadata to
-	// be enabled as well - this is to allow for key-rotation.
-	// TODO: something need to Encrypted with SSE
-	//if directive != "REPLACE" && isMetadataOnly && !IsEncrypted(targetObject.CustomAttributes) {
-	if directive != "REPLACE" && isMetadataOnly {
-		WriteErrorResponse(w, r, ErrInvalidCopyRequestWithSameObject)
-		return
-	}
-
 	// Create the object.
 	result, err := api.ObjectAPI.CopyObject(reqCtx, targetObject, pipeReader, credential, sseRequest, isMetadataOnly)
 	if err != nil {

collector.go

@@ -55,24 +55,28 @@ func (c *Metrics) Collect(ch chan<- prometheus.Metric) {
 
 	GaugeMetricDataForBucket := c.GenerateBucketUsageData()
 	for bucket, data := range GaugeMetricDataForBucket {
-		ch <- prometheus.MustNewConstMetric(c.metrics["bucket_usage_byte_metric"], prometheus.GaugeValue, float64(data.value), bucket, data.owner, data.storageClass)
+		for _, v := range data {
+			ch <- prometheus.MustNewConstMetric(c.metrics["bucket_usage_byte_metric"], prometheus.GaugeValue, float64(v.value), bucket, v.owner, v.storageClass)
+		}
 	}
 
 	GaugeMetricDataForUid := c.GenerateUserUsageData()
 	for uid, data := range GaugeMetricDataForUid {
-		ch <- prometheus.MustNewConstMetric(c.metrics["user_usage_byte_metric"], prometheus.GaugeValue, float64(data.value), uid, data.storageClass)
+		for _, v := range data {
+			ch <- prometheus.MustNewConstMetric(c.metrics["user_usage_byte_metric"], prometheus.GaugeValue, float64(v.value), uid, v.storageClass)
+		}
 	}
 }
 
 // Get bucket usage cache which like <key><value> = <u_b_test><STANDARD:233333>
-func (c *Metrics) GenerateBucketUsageData() (GaugeMetricData map[string]UsageDataWithBucket) {
+func (c *Metrics) GenerateBucketUsageData() (GaugeMetricData map[string][]UsageDataWithBucket) {
 	buckets, err := adminServer.Yig.MetaStorage.GetBuckets()
 	if err != nil {
 		helper.Logger.Error("Get usage data for prometheus failed:",
 			err.Error())
 		return
 	}
-	GaugeMetricData = make(map[string]UsageDataWithBucket)
+	GaugeMetricData = make(map[string][]UsageDataWithBucket)
 	for _, bucket := range buckets {
 		key := BucketUsagePrefix + bucket.Name
 		usageCache, err := redis.GetUsage(key)
@@ -81,65 +85,68 @@ func (c *Metrics) GenerateBucketUsageData() (GaugeMetricData map[string]UsageDat
 				err.Error())
 			return
 		}
-		data, err := parseUsage(usageCache)
+		datas, err := parseUsage(usageCache)
 		if err != nil {
 			helper.Logger.Error("Parse usage data from redis for prometheus failed:",
 				err.Error())
 			return
 		}
-		if data.storageClass != "" {
-			GaugeMetricData[bucket.Name] = UsageDataWithBucket{data.value, bucket.OwnerId, data.storageClass}
+		for _, data := range datas {
+			GaugeMetricData[bucket.Name] = append(GaugeMetricData[bucket.Name], UsageDataWithBucket{data.value, bucket.OwnerId, data.storageClass})
 		}
 	}
 	return
 }
 
 // Get bucket usage cache which like <key><value> = <u_p_hehehehe><STANDARD 233333>
-func (c *Metrics) GenerateUserUsageData() (GaugeMetricData map[string]UsageData) {
+func (c *Metrics) GenerateUserUsageData() (GaugeMetricData map[string][]UsageData) {
 	buckets, err := adminServer.Yig.MetaStorage.GetBuckets()
 	if err != nil {
 		helper.Logger.Error("Get usage data for prometheus failed:",
 			err.Error())
 		return
 	}
-	GaugeMetricData = make(map[string]UsageData)
+	GaugeMetricData = make(map[string][]UsageData)
 	for _, bucket := range buckets {
-		if GaugeMetricData[bucket.OwnerId].storageClass == "" {
+		if len(GaugeMetricData[bucket.OwnerId]) == 0 {
 			key := PidUsagePrefix + bucket.OwnerId
 			usageCache, err := redis.GetUsage(key)
 			if err != nil {
 				helper.Logger.Error("Get usage data from redis for prometheus failed:",
 					err.Error())
 				return
 			}
-			data, err := parseUsage(usageCache)
+			datas, err := parseUsage(usageCache)
 			if err != nil {
 				helper.Logger.Error("Parse usage data from redis for prometheus failed:",
 					err.Error())
 				return
 			}
-			if data.storageClass != "" {
-				GaugeMetricData[bucket.OwnerId] = UsageData{value: data.value, storageClass: data.storageClass}
+			for _, data := range datas {
+				GaugeMetricData[bucket.OwnerId] = append(GaugeMetricData[bucket.OwnerId], UsageData{value: data.value, storageClass: data.storageClass})
 			}
 		}
 	}
 	return
 }
 
 //  get usage from redis
-//  <Storage-Class>:<usagenumber>
+//  <Storage-Class1>:<usagenumber>,<Storage-Class2>:<usagenumber>
 //  eg. STANDARD:2222
-func parseUsage(value string) (*UsageData, error) {
-	var err error
-	data := new(UsageData)
+func parseUsage(value string) (datas []*UsageData, err error) {
 	if value == "" {
-		return data, nil
+		return
 	}
-	allParams := strings.Split(value, ":")
-	data.value, err = strconv.ParseInt(allParams[1], 10, 64)
-	if err != nil {
-		return data, err
+	storageClass := strings.Split(value, ",")
+	for _, v := range storageClass {
+		data := new(UsageData)
+		allParams := strings.Split(v, ":")
+		data.value, err = strconv.ParseInt(allParams[1], 10, 64)
+		if err != nil {
+			return
+		}
+		data.storageClass = allParams[0]
+		datas = append(datas, data)
 	}
-	data.storageClass = allParams[0]
-	return data, nil
+	return
 }

conf/yig.toml

@@ -23,6 +23,7 @@ meta_cache_type = 2
 meta_store = "tidb"
 tidb_info = "root:@tcp(10.5.0.17:4000)/yig"
 keepalive = true
+enable_usage_push = false
 redis_group = ["redis:6379"]
 redis_password = "hehehehe"
 redis_connection_number = 10
@@ -61,15 +62,13 @@ kms_secret = "your_secret"
 version = 0
 keyname = "yig"
 
-[msg_bus]
-msg_bus_enable = true
-msg_bus_type = 1
-msg_bus_topic = "testTopic2"
-[msg_bus.msg_bus_server]
-broker_list = "kafka:29092"
-
-
 # Plugin Config
+[plugins.dummy_mq]
+path = "/etc/yig/plugins/dummy_mq_plugin.so"
+enable = true
+[plugins.dummy_mq.args]
+topic = "testTopic2"
+url = "kafka:29092"
 
 [plugins.dummy_iam]
 path = "/etc/yig/plugins/dummy_iam_plugin.so"

helper/config.go

@@ -80,9 +80,6 @@ type Config struct {
 	UploadMaxChunkSize  int64 `toml:"upload_max_chunk_size"`
 
 	KMS KMSConfig `toml:"kms"`
-
-	// Message Bus
-	MsgBus MsgBusConfig `toml:"msg_bus"`
 }
 
 type PluginConfig struct {
@@ -100,25 +97,6 @@ type KMSConfig struct {
 	Keyname  string
 }
 
-type MsgBusConfig struct {
-	// Controls whether to enable message bus when receive the request.
-	Enabled bool `toml:"msg_bus_enable"`
-	// Controls the under implementation of message bus: 1 for kafka.
-	Type int `toml:"msg_bus_type"`
-	// Controls the message topic used by message bus.
-	Topic string `toml:"msg_bus_topic"`
-	// Controls the request timeout for sending a req through message bus.
-	RequestTimeoutMs int `toml:"msg_bus_request_timeout_ms"`
-	// Controls the total timeout for sending a req through message bus.
-	// It will timeout if min(MessageTimeoutMs, SendMaxRetries * RequestTimeoutMs) meets.
-	MessageTimeoutMs int `toml:"msg_bus_message_timeout_ms"`
-	// Controls the retry time used by message bus if it fails to send a req.
-	SendMaxRetries int `toml:"msg_bus_send_max_retries"`
-	// Controls the settings for the implementation of message bus.
-	// For kafka, the 'broker_list' must be set, like 'broker_list = "kafka:29092"'
-	Server map[string]interface{} `toml:"msg_bus_server"`
-}
-
 var CONFIG Config
 
 func SetupConfig() {
@@ -203,10 +181,5 @@ func MarshalTOMLConfig() error {
 
 	CONFIG.KMS = c.KMS
 
-	CONFIG.MsgBus = c.MsgBus
-	CONFIG.MsgBus.RequestTimeoutMs = Ternary(c.MsgBus.RequestTimeoutMs == 0, 3000, c.MsgBus.RequestTimeoutMs).(int)
-	CONFIG.MsgBus.MessageTimeoutMs = Ternary(c.MsgBus.MessageTimeoutMs == 0, 5000, c.MsgBus.MessageTimeoutMs).(int)
-	CONFIG.MsgBus.SendMaxRetries = Ternary(c.MsgBus.SendMaxRetries == 0, 2, c.MsgBus.SendMaxRetries).(int)
-
 	return nil
 }