You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
minio/cmd/notify-elasticsearch.go

159 lines
3.8 KiB

/*
* 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 cmd
import (
"encoding/hex"
"errors"
"fmt"
"io/ioutil"
"github.com/Sirupsen/logrus"
"github.com/minio/sha256-simd"
"gopkg.in/olivere/elastic.v3"
)
// elasticQueue is a elasticsearch event notification queue.
type elasticSearchNotify struct {
Enable bool `json:"enable"`
Format string `json:"format"`
URL string `json:"url"`
Index string `json:"index"`
}
func (e *elasticSearchNotify) Validate() error {
if !e.Enable {
return nil
}
if e.Format != formatNamespace {
return fmt.Errorf(
"Elasticsearch Notifier Error: \"format\" must be \"%s\"",
formatNamespace)
}
if _, err := checkURL(e.URL); err != nil {
return err
}
return nil
}
type elasticClient struct {
*elastic.Client
params elasticSearchNotify
}
// Connects to elastic search instance at URL.
func dialElastic(esNotify elasticSearchNotify) (*elastic.Client, error) {
if !esNotify.Enable {
return nil, errNotifyNotEnabled
}
client, err := elastic.NewClient(
elastic.SetURL(esNotify.URL),
elastic.SetSniff(false),
elastic.SetMaxRetries(10),
)
if err != nil {
return nil, err
}
return client, nil
}
func newElasticNotify(accountID string) (*logrus.Logger, error) {
esNotify := serverConfig.Notify.GetElasticSearchByID(accountID)
// Dial to elastic search.
client, err := dialElastic(esNotify)
if err != nil {
return nil, err
}
// Use the IndexExists service to check if a specified index exists.
exists, err := client.IndexExists(esNotify.Index).Do()
if err != nil {
return nil, err
}
// Index does not exist, attempt to create it.
if !exists {
var createIndex *elastic.IndicesCreateResult
createIndex, err = client.CreateIndex(esNotify.Index).Do()
if err != nil {
return nil, err
}
if !createIndex.Acknowledged {
return nil, errors.New("Index not created")
}
}
elasticCl := elasticClient{
Client: client,
params: esNotify,
}
elasticSearchLog := logrus.New()
// Disable writing to console.
elasticSearchLog.Out = ioutil.Discard
// Add a elasticSearch hook.
elasticSearchLog.Hooks.Add(elasticCl)
// Set default JSON formatter.
elasticSearchLog.Formatter = new(logrus.JSONFormatter)
// Success, elastic search successfully initialized.
return elasticSearchLog, nil
}
// Fire is required to implement logrus hook
func (q elasticClient) Fire(entry *logrus.Entry) error {
// Reflect on eventType and Key on their native type.
entryStr, ok := entry.Data["EventType"].(string)
if !ok {
return nil
}
keyStr, ok := entry.Data["Key"].(string)
if !ok {
return nil
}
// Calculate a unique key id. Choosing sha256 here.
shaKey := sha256.Sum256([]byte(keyStr))
keyStr = hex.EncodeToString(shaKey[:])
// If event matches as delete, we purge the previous index.
if eventMatch(entryStr, []string{"s3:ObjectRemoved:*"}) {
_, err := q.Client.Delete().Index(q.params.Index).
Type("event").Id(keyStr).Do()
if err != nil {
return err
}
return nil
} // else we update elastic index or create a new one.
_, err := q.Client.Index().Index(q.params.Index).
Type("event").
BodyJson(map[string]interface{}{
"Records": entry.Data["Records"],
}).Id(keyStr).Do()
return err
}
// Required for logrus hook implementation
func (q elasticClient) Levels() []logrus.Level {
return []logrus.Level{
logrus.InfoLevel,
}
}