minio/queues.go

/*
 * Minio Cloud Storage, (C) 2016 Minio, Inc.
 *
 * 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 main

import (
	"fmt"
	"net/url"
	"time"

	"github.com/Sirupsen/logrus"
)

const (
	minioSqs = "arn:minio:sqs:"
	// Static string indicating queue type 'amqp'.
	queueTypeAMQP = "1:amqp"
	// Static string indicating queue type 'elasticsearch'.
	queueTypeElastic = "1:elasticsearch"
	// Static string indicating queue type 'redis'.
	queueTypeRedis = "1:redis"
)

// Returns true if queueArn is for an AMQP queue.
func isAMQPQueue(sqsArn arnMinioSqs) bool {
	if sqsArn.sqsType == queueTypeAMQP {
		amqpL := serverConfig.GetAMQPLogger()
		if !amqpL.Enable {
			return false
		}
		// Connect to amqp server to validate.
		amqpC, err := dialAMQP(amqpL)
		if err != nil {
			errorIf(err, "Unable to connect to amqp service.", amqpL)
			return false
		}
		defer amqpC.Close()
	}
	return true
}

// Returns true if queueArn is for an Redis queue.
func isRedisQueue(sqsArn arnMinioSqs) bool {
	if sqsArn.sqsType == queueTypeRedis {
		rLogger := serverConfig.GetRedisLogger()
		if !rLogger.Enable {
			return false
		}
		// Connect to redis server to validate.
		rPool, err := dialRedis(rLogger.Addr, rLogger.Password)
		if err != nil {
			errorIf(err, "Unable to connect to redis service.", rLogger)
			return false
		}
		defer rPool.Close()
	}
	return true
}

// Returns true if queueArn is for an ElasticSearch queue.
func isElasticQueue(sqsArn arnMinioSqs) bool {
	if sqsArn.sqsType == queueTypeElastic {
		esLogger := serverConfig.GetElasticSearchLogger()
		if !esLogger.Enable {
			return false
		}
		elasticC, err := dialElastic(esLogger.URL)
		if err != nil {
			errorIf(err, "Unable to connect to elasticsearch service.", esLogger.URL)
			return false
		}
		defer elasticC.Stop()
	}
	return true
}

// Match function matches wild cards in 'pattern' for events.
func eventMatch(eventType EventName, events []string) (ok bool) {
	for _, event := range events {
		ok = wildCardMatch(event, eventType.String())
		if ok {
			break
		}
	}
	return ok
}

// NotifyObjectCreatedEvent - notifies a new 's3:ObjectCreated' event.
// List of events reported through this function are
//  - s3:ObjectCreated:Put
//  - s3:ObjectCreated:Post
//  - s3:ObjectCreated:Copy
//  - s3:ObjectCreated:CompleteMultipartUpload
func notifyObjectCreatedEvent(nConfig notificationConfig, eventType EventName, bucket string, object string, etag string, size int64) {
	/// Construct a new object created event.
	region := serverConfig.GetRegion()
	tnow := time.Now().UTC()
	sequencer := fmt.Sprintf("%X", tnow.UnixNano())
	// Following blocks fills in all the necessary details of s3 event message structure.
	// http://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
	events := []*NotificationEvent{
		{
			EventVersion:      "2.0",
			EventSource:       "aws:s3",
			AwsRegion:         region,
			EventTime:         tnow.Format(iso8601Format),
			EventName:         eventType.String(),
			UserIdentity:      defaultIdentity(),
			RequestParameters: map[string]string{},
			ResponseElements:  map[string]string{},
			S3: s3Reference{
				SchemaVersion:   "1.0",
				ConfigurationID: "Config",
				Bucket: s3BucketReference{
					Name:          bucket,
					OwnerIdentity: defaultIdentity(),
					ARN:           "arn:aws:s3:::" + bucket,
				},
				Object: s3ObjectReference{
					Key:       url.QueryEscape(object),
					ETag:      etag,
					Size:      size,
					Sequencer: sequencer,
				},
			},
		},
	}
	// Notify to all the configured queues.
	for _, qConfig := range nConfig.QueueConfigurations {
		if eventMatch(eventType, qConfig.Events) {
			log.WithFields(logrus.Fields{
				"Records": events,
			}).Info()
		}
	}
}

// NotifyObjectRemovedEvent - notifies a new 's3:ObjectRemoved' event.
// List of events reported through this function are
//  - s3:ObjectRemoved:Delete
func notifyObjectDeletedEvent(nConfig notificationConfig, bucket string, object string) {
	region := serverConfig.GetRegion()
	tnow := time.Now().UTC()
	sequencer := fmt.Sprintf("%X", tnow.UnixNano())
	// Following blocks fills in all the necessary details of s3 event message structure.
	// http://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
	events := []*NotificationEvent{
		{
			EventVersion:      "2.0",
			EventSource:       "aws:s3",
			AwsRegion:         region,
			EventTime:         tnow.Format(iso8601Format),
			EventName:         ObjectRemovedDelete.String(),
			UserIdentity:      defaultIdentity(),
			RequestParameters: map[string]string{},
			ResponseElements:  map[string]string{},
			S3: s3Reference{
				SchemaVersion:   "1.0",
				ConfigurationID: "Config",
				Bucket: s3BucketReference{
					Name:          bucket,
					OwnerIdentity: defaultIdentity(),
					ARN:           "arn:aws:s3:::" + bucket,
				},
				Object: s3ObjectReference{
					Key:       url.QueryEscape(object),
					Sequencer: sequencer,
				},
			},
		},
	}
	// Notify to all the configured queues.
	for _, qConfig := range nConfig.QueueConfigurations {
		if eventMatch(ObjectRemovedDelete, qConfig.Events) {
			log.WithFields(logrus.Fields{
				"Records": events,
			}).Info()
		}
	}
}