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-postgresql.go

253 lines
6.6 KiB

/*
* Minio Cloud Storage, (C) 2014-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.
*/
// PostgreSQL Notifier implementation. A table with a specific
// structure (column names, column types, and primary key/uniqueness
// constraint) is used. The user may set the table name in the
// configuration. A sample SQL command that creates a command with the
// required structure is:
//
// CREATE TABLE myminio (
// key VARCHAR PRIMARY KEY,
// value JSONB
// );
//
// PostgreSQL's "INSERT ... ON CONFLICT ... DO UPDATE ..." feature
// (UPSERT) is used here, so the minimum version of PostgreSQL
// required is 9.5.
//
// On each create or update object event in Minio Object storage
// server, a row is created or updated in the table in Postgres. On
// each object removal, the corresponding row is deleted from the
// table.
package cmd
import (
"database/sql"
"encoding/json"
"fmt"
"io/ioutil"
"strings"
"github.com/Sirupsen/logrus"
// libpq db driver is usually imported blank - see examples in
// https://godoc.org/github.com/lib/pq
_ "github.com/lib/pq"
)
const (
upsertRow = `INSERT INTO %s (key, value)
VALUES ($1, $2)
ON CONFLICT (key)
DO UPDATE SET value = EXCLUDED.value;`
deleteRow = ` DELETE FROM %s
WHERE key = $1;`
createTable = `CREATE TABLE %s (
key VARCHAR PRIMARY KEY,
value JSONB
);`
tableExists = `SELECT 1 FROM %s;`
)
type postgreSQLNotify struct {
Enable bool `json:"enable"`
// pass connection string in config directly. This string is
// formatted according to
// https://godoc.org/github.com/lib/pq#hdr-Connection_String_Parameters
ConnectionString string `json:"connectionString"`
// specifying a table name is required.
Table string `json:"table"`
// uses the values below if no connection string is specified
// - however the connection string method offers more
// flexibility.
Host string `json:"host"`
Port string `json:"port"`
User string `json:"user"`
Password string `json:"password"`
Database string `json:"database"`
}
type pgConn struct {
connStr string
table string
preparedStmts map[string]*sql.Stmt
*sql.DB
}
func dialPostgreSQL(pgN postgreSQLNotify) (pgConn, error) {
if !pgN.Enable {
return pgConn{}, errNotifyNotEnabled
}
// check that table is specified
if pgN.Table == "" {
return pgConn{}, fmt.Errorf(
"PostgreSQL Notifier Error: Table was not specified in configuration")
}
connStr := pgN.ConnectionString
// check if connection string is specified
if connStr == "" {
// build from other parameters
params := []string{}
if pgN.Host != "" {
params = append(params, "host="+pgN.Host)
}
if pgN.Port != "" {
params = append(params, "port="+pgN.Port)
}
if pgN.User != "" {
params = append(params, "user="+pgN.User)
}
if pgN.Password != "" {
params = append(params, "password="+pgN.Password)
}
if pgN.Database != "" {
params = append(params, "dbname="+pgN.Database)
}
connStr = strings.Join(params, " ")
}
db, err := sql.Open("postgres", connStr)
if err != nil {
return pgConn{}, fmt.Errorf(
"PostgreSQL Notifier Error: Connection opening failure (connectionString=%s): %v",
connStr, err,
)
}
// ping to check that server is actually reachable.
err = db.Ping()
if err != nil {
return pgConn{}, fmt.Errorf(
"PostgreSQL Notifier Error: Ping to server failed with: %v",
err,
)
}
// check that table exists - if not, create it.
_, err = db.Exec(fmt.Sprintf(tableExists, pgN.Table))
if err != nil {
// most likely, table does not exist. try to create it:
_, errCreate := db.Exec(fmt.Sprintf(createTable, pgN.Table))
if errCreate != nil {
// failed to create the table. error out.
return pgConn{}, fmt.Errorf(
"PostgreSQL Notifier Error: 'Select' failed with %v, then 'Create Table' failed with %v",
err, errCreate,
)
}
}
// create prepared statements
stmts := make(map[string]*sql.Stmt)
// insert or update statement
stmts["upsertRow"], err = db.Prepare(fmt.Sprintf(upsertRow, pgN.Table))
if err != nil {
return pgConn{},
fmt.Errorf("PostgreSQL Notifier Error: create UPSERT prepared statement failed with: %v", err)
}
stmts["deleteRow"], err = db.Prepare(fmt.Sprintf(deleteRow, pgN.Table))
if err != nil {
return pgConn{},
fmt.Errorf("PostgreSQL Notifier Error: create DELETE prepared statement failed with: %v", err)
}
return pgConn{connStr, pgN.Table, stmts, db}, nil
}
func newPostgreSQLNotify(accountID string) (*logrus.Logger, error) {
pgNotify := serverConfig.GetPostgreSQLNotifyByID(accountID)
// Dial postgres
pgC, err := dialPostgreSQL(pgNotify)
if err != nil {
return nil, err
}
pgLog := logrus.New()
pgLog.Out = ioutil.Discard
pgLog.Formatter = new(logrus.JSONFormatter)
pgLog.Hooks.Add(pgC)
return pgLog, nil
}
func (pgC pgConn) Close() {
// first close all prepared statements
for _, v := range pgC.preparedStmts {
_ = v.Close()
}
// close db connection
_ = pgC.DB.Close()
}
func (pgC pgConn) Fire(entry *logrus.Entry) error {
// get event type by trying to convert to string
entryEventType, ok := entry.Data["EventType"].(string)
if !ok {
// ignore event if converting EventType to string
// fails.
return nil
}
// Check for event delete
if eventMatch(entryEventType, []string{"s3:ObjectRemoved:*"}) {
// delete row from the table
_, err := pgC.preparedStmts["deleteRow"].Exec(entry.Data["Key"])
if err != nil {
return fmt.Errorf(
"Error deleting event with key = %v - got postgres error - %v",
entry.Data["Key"], err,
)
}
} else {
// json encode the value for the row
value, err := json.Marshal(map[string]interface{}{
"Records": entry.Data["Records"],
})
if err != nil {
return fmt.Errorf(
"Unable to encode event %v to JSON - got error - %v",
entry.Data["Records"], err,
)
}
// upsert row into the table
_, err = pgC.preparedStmts["upsertRow"].Exec(entry.Data["Key"], value)
if err != nil {
return fmt.Errorf(
"Unable to upsert event with Key=%v and Value=%v - got postgres error - %v",
entry.Data["Key"], entry.Data["Records"], err,
)
}
}
return nil
}
func (pgC pgConn) Levels() []logrus.Level {
return []logrus.Level{
logrus.InfoLevel,
}
}