/* ** Zabbix ** Copyright (C) 2001-2023 Zabbix SIA ** ** This program is free software; you can redistribute it and/or modify ** it under the terms of the GNU General Public License as published by ** the Free Software Foundation; either version 2 of the License, or ** (at your option) any later version. ** ** This program is distributed in the hope that it will be useful, ** but WITHOUT ANY WARRANTY; without even the implied warranty of ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ** GNU General Public License for more details. ** ** You should have received a copy of the GNU General Public License ** along with this program; if not, write to the Free Software ** Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ package oracle import ( "context" "database/sql" "fmt" "net/url" "strings" "sync" "time" "git.zabbix.com/ap/plugin-support/uri" "git.zabbix.com/ap/plugin-support/log" "git.zabbix.com/ap/plugin-support/zbxerr" "github.com/godror/godror" "github.com/omeid/go-yarn" ) type OraClient interface { Query(ctx context.Context, query string, args ...interface{}) (rows *sql.Rows, err error) QueryByName(ctx context.Context, queryName string, args ...interface{}) (rows *sql.Rows, err error) QueryRow(ctx context.Context, query string, args ...interface{}) (row *sql.Row, err error) QueryRowByName(ctx context.Context, queryName string, args ...interface{}) (row *sql.Row, err error) WhoAmI() string } type OraConn struct { client *sql.DB callTimeout time.Duration version godror.VersionInfo lastTimeAccess time.Time ctx context.Context queryStorage *yarn.Yarn username string } var errorQueryNotFound = "query %q not found" // Query wraps DB.QueryContext. func (conn *OraConn) Query(ctx context.Context, query string, args ...interface{}) (rows *sql.Rows, err error) { rows, err = conn.client.QueryContext(ctx, query, args...) if ctxErr := ctx.Err(); ctxErr != nil { err = ctxErr } return } // Query executes a query from queryStorage by its name and returns multiple rows. func (conn *OraConn) QueryByName(ctx context.Context, queryName string, args ...interface{}) (rows *sql.Rows, err error) { if sql, ok := (*conn.queryStorage).Get(queryName + sqlExt); ok { normalizedSQL := strings.TrimRight(strings.TrimSpace(sql), ";") return conn.Query(ctx, normalizedSQL, args...) } return nil, fmt.Errorf(errorQueryNotFound, queryName) } // Query wraps DB.QueryRowContext. func (conn *OraConn) QueryRow(ctx context.Context, query string, args ...interface{}) (row *sql.Row, err error) { row = conn.client.QueryRowContext(ctx, query, args...) if ctxErr := ctx.Err(); ctxErr != nil { err = ctxErr } return } // Query executes a query from queryStorage by its name and returns a single row. func (conn *OraConn) QueryRowByName(ctx context.Context, queryName string, args ...interface{}) (row *sql.Row, err error) { if sql, ok := (*conn.queryStorage).Get(queryName + sqlExt); ok { normalizedSQL := strings.TrimRight(strings.TrimSpace(sql), ";") return conn.QueryRow(ctx, normalizedSQL, args...) } return nil, fmt.Errorf(errorQueryNotFound, queryName) } // WhoAmI returns a current username. func (conn *OraConn) WhoAmI() string { return conn.username } // updateAccessTime updates the last time a connection was accessed. func (conn *OraConn) updateAccessTime() { conn.lastTimeAccess = time.Now() } // ConnManager is thread-safe structure for manage connections. type ConnManager struct { sync.Mutex connMutex sync.Mutex connections map[connDetails]*OraConn keepAlive time.Duration connectTimeout time.Duration callTimeout time.Duration Destroy context.CancelFunc queryStorage yarn.Yarn } type connDetails struct { uri uri.URI privilege string } // NewConnManager initializes connManager structure and runs Go Routine that watches for unused connections. func NewConnManager(keepAlive, connectTimeout, callTimeout, hkInterval time.Duration, queryStorage yarn.Yarn) *ConnManager { ctx, cancel := context.WithCancel(context.Background()) connMgr := &ConnManager{ connections: make(map[connDetails]*OraConn), keepAlive: keepAlive, connectTimeout: connectTimeout, callTimeout: callTimeout, Destroy: cancel, // Destroy stops originated goroutines and closes connections. queryStorage: queryStorage, } go connMgr.housekeeper(ctx, hkInterval) return connMgr } // closeUnused closes each connection that has not been accessed at least within the keepalive interval. func (c *ConnManager) closeUnused() { c.connMutex.Lock() defer c.connMutex.Unlock() for cd, conn := range c.connections { if time.Since(conn.lastTimeAccess) > c.keepAlive { conn.client.Close() delete(c.connections, cd) log.Debugf("[%s] Closed unused connection: %s", pluginName, cd.uri.Addr()) } } } // closeAll closes all existed connections. func (c *ConnManager) closeAll() { c.connMutex.Lock() for uri, conn := range c.connections { conn.client.Close() delete(c.connections, uri) } c.connMutex.Unlock() } // housekeeper repeatedly checks for unused connections and closes them. func (c *ConnManager) housekeeper(ctx context.Context, interval time.Duration) { ticker := time.NewTicker(interval) for { select { case <-ctx.Done(): ticker.Stop() c.closeAll() return case <-ticker.C: c.closeUnused() } } } // create creates a new connection with given credentials. func (c *ConnManager) create(cd connDetails) (*OraConn, error) { c.connMutex.Lock() defer c.connMutex.Unlock() if _, ok := c.connections[cd]; ok { // Should never happen. panic("connection already exists") } ctx := godror.ContextWithTraceTag( context.Background(), godror.TraceTag{ ClientInfo: "zbx_monitor", Module: godror.DriverName, }) service, err := url.QueryUnescape(cd.uri.GetParam("service")) if err != nil { return nil, err } connectString := fmt.Sprintf(`(DESCRIPTION=(ADDRESS=(PROTOCOL=tcp)(HOST=%s)(PORT=%s))`+ `(CONNECT_DATA=(SERVICE_NAME="%s"))(CONNECT_TIMEOUT=%d)(RETRY_COUNT=0))`, cd.uri.Host(), cd.uri.Port(), service, c.connectTimeout/time.Second) connParams, err := getConnParams(cd.privilege) if err != nil { return nil, zbxerr.ErrorInvalidParams.Wrap(err) } connector := godror.NewConnector(godror.ConnectionParams{ StandaloneConnection: true, CommonParams: godror.CommonParams{ Username: cd.uri.User(), ConnectString: connectString, Password: godror.NewPassword(cd.uri.Password()), }, ConnParams: connParams, }) client := sql.OpenDB(connector) serverVersion, err := godror.ServerVersion(ctx, client) if err != nil { return nil, err } c.connections[cd] = &OraConn{ client: client, callTimeout: c.callTimeout, version: serverVersion, lastTimeAccess: time.Now(), ctx: ctx, queryStorage: &c.queryStorage, username: cd.uri.User(), } log.Debugf("[%s] Created new connection: %s", pluginName, cd.uri.Addr()) return c.connections[cd], nil } // get returns a connection with given uri if it exists and also updates lastTimeAccess, otherwise returns nil. func (c *ConnManager) get(cd connDetails) *OraConn { c.connMutex.Lock() defer c.connMutex.Unlock() if conn, ok := c.connections[cd]; ok { conn.updateAccessTime() return conn } return nil } // GetConnection returns an existing connection or creates a new one. func (c *ConnManager) GetConnection(cd connDetails) (conn *OraConn, err error) { c.Lock() defer c.Unlock() conn = c.get(cd) if conn == nil { conn, err = c.create(cd) } if err != nil { if oraErr, isOraErr := godror.AsOraErr(err); isOraErr { err = zbxerr.ErrorConnectionFailed.Wrap(oraErr) } else { err = zbxerr.ErrorConnectionFailed.Wrap(err) } } return } func getConnParams(privilege string) (out godror.ConnParams, err error) { switch privilege { case "sysdba": out.IsSysDBA = true case "sysoper": out.IsSysOper = true case "sysasm": out.IsSysASM = true case "": default: err = fmt.Errorf("incorrect privilege, %s", privilege) return } return }