changes for adding statement cache support and sql_id generation

cal/calconfig.go

@@ -108,7 +108,7 @@ func (c *calConfig) initialize(cfg config.Config, vcfg config.Config, _labelAffi
 	c.msgBufferSize = cfg.GetOrDefaultInt("cal_message_buffer_size", 300)
 	c.handlerType = cfg.GetOrDefaultString("cal_handler", "socket")
 	c.logFileName = cfg.GetOrDefaultString("cal_log_file", "logCalClient.txt")
-	c.enableTG = (cfg.GetOrDefaultString("cal_enable_threadgroup", "false") == "true")
+	c.enableTG = (cfg.GetOrDefaultString("cal_enable_threadgroup", "true") == "true")
 	c.poolstackEnabled = (cfg.GetOrDefaultString("cal_pool_stack_enable", "true") == "true")
 	c.poolStackSize = cfg.GetOrDefaultInt("cal_max_pool_stack_size", 2048)
 	if c.poolStackSize > 2048 {

cal/calinterface.go

@@ -63,6 +63,12 @@ type Transaction interface {
 	SetOperationName(string, bool)
 }
 
+// AtomicTransaction declares the functions for generating CAL transactions
+type AtomicTransaction interface {
+	Activity
+	SetDuration(float32)
+}
+
 // HeartBeat declares the functions for generating CAL heartbeat
 type HeartBeat interface {
 	Activity

cal/calmessage.go

@@ -171,6 +171,12 @@ type calTransaction struct {
 	mTimer CalTimer
 }
 
+type calAtomicTransaction struct {
+	calActivity
+	mDuration float32
+	mTimer CalTimer
+}
+
 // NewCalEvent creates a CAL event
 func NewCalEvent(_type string, _name string, _status string, _data string, _tgname ...string) Event {
 	et := new(calEvent)
@@ -200,6 +206,13 @@ func NewCalTransaction(_type string, _name string, _status string, _data string,
 	return ct
 }
 
+// NewCalAtomicTransaction creates a CAL atomic transaction
+func NewCalAtomicTransaction(_type string, _name string, _status string, _data string, _tgname string) AtomicTransaction {
+	ct := new(calAtomicTransaction)
+	ct.init(_type, _name, _status, _data, _tgname)
+	return ct
+}
+
 // ReleaseCxtResource releases the internal context
 func ReleaseCxtResource(_tgname ...string) {
 	var client = GetCalClientInstance()
@@ -723,6 +736,80 @@ func (act *calHeartBeat) init(_type string, _name string, _status string, _data
 	}
 }
 
+/********************************************************
+ ********************************************************
+ *               calAtomicTransaction
+ ********************************************************
+ */
+ func (act *calAtomicTransaction) init(_type string, _name string, _status string, _data string, _tgname string) {
+	if !act.isCalClientEnabled() {
+		return
+	}
+	act.mClass = calClassAtomicTransaction
+	act.initialize(_type, _name, _status, _data, _tgname)
+
+	act.mTimer.Reset()
+	act.mDuration = -1
+
+}
+
+func (act *calAtomicTransaction) SetDuration(_duration float32) {
+	if !act.isCalClientEnabled() {
+		return
+	}
+	if _duration < minDuration {
+		act.mDuration = minDuration
+	} else if _duration > maxDuration {
+		act.mDuration = maxDuration
+	} else {
+		act.mDuration = _duration
+	}
+}
+
+func (act *calAtomicTransaction) Completed() {
+	if act.mCompleted {
+		return
+	}
+	if !act.isCalClientEnabled() {
+		return
+	}
+
+	act.sendSelf()
+	act.mCompleted = true
+}
+
+func (act *calAtomicTransaction) sendSelf() {
+	var duration_str string
+	// to safeguard 64 to 32 bit int conversion
+	value := act.mTimer.Duration()
+	var duration float32
+	if int32(value) > maxDuration {		// Check on comparision between float and int
+		duration = maxDuration
+	} else {
+		duration = value
+	}
+	if act.mDuration >= minDuration{
+		duration = act.mDuration
+	}
+	duration_str = fmt.Sprintf("%.2f", duration)
+	var buf bytes.Buffer
+	buf.WriteString(act.mClass)
+	buf.WriteString(act.mTimeStamp)
+	buf.WriteString(calTab)
+	buf.WriteString(act.mType)
+	buf.WriteString(calTab)
+	buf.WriteString(act.mName)
+	buf.WriteString(calTab)
+	buf.WriteString(act.mStatus)
+	buf.WriteString(calTab)
+	buf.WriteString(duration_str)
+	buf.WriteString(calTab)
+	buf.WriteString(act.mData)
+	buf.WriteString(calEndOfLine)
+	var str = buf.String()
+	act.writeData(str)
+}
+
 /********************************************************
  ********************************************************
  *               calTransaction
@@ -1053,10 +1140,10 @@ func (act *calTransaction) completeAnyNestedTransactions() {
  */
 func (act *calTransaction) addAdditionalFieldsForRoot() {
 	var buf bytes.Buffer
-	buf.WriteString("corr_id_")
-	buf.WriteString(calEquals)
-	buf.WriteString(act.GetCorrelationID())
-	buf.WriteString(calAmpersand)
+	// buf.WriteString("corr_id_")
+	// buf.WriteString(calEquals)
+	// buf.WriteString(act.GetCorrelationID())
+	// buf.WriteString(calAmpersand)
 	buf.WriteString("log_id_")
 	buf.WriteString(calEquals)
 	//buf.WriteString(client.GetLogId())

cal/caltimer.go

@@ -40,9 +40,9 @@ func (ct *CalTimer) Reset(){
   ct.tBegin = time.Now()
 }
 
-// return time elapsed since last Reset call in 10th of micro seconds
+// return time elapsed since last Reset call in milliseconds
 func (ct *CalTimer) Duration() float32{
   var timediff time.Duration
   timediff = time.Since(ct.tBegin)
-  return float32(timediff.Nanoseconds()) / float32(time.Millisecond * 10)
+  return float32(timediff.Nanoseconds()) / float32(time.Millisecond)
 }

client/gosqldriver/connection.go

@@ -22,12 +22,11 @@ import (
 	"database/sql/driver"
 	"errors"
 	"fmt"
-	"net"
-	"os"
-
 	"github.com/paypal/hera/common"
 	"github.com/paypal/hera/utility/encoding/netstring"
 	"github.com/paypal/hera/utility/logger"
+	"net"
+	"os"
 )
 
 var corrIDUnsetCmd = netstring.NewNetstringFrom(common.CmdClientCalCorrelationID, []byte("CorrId=NotSet"))
@@ -38,6 +37,10 @@ type heraConnection struct {
 	reader *netstring.Reader
 	// for the sharding extension
 	shardKeyPayload []byte
+	//caching extension
+	cacheKey []byte
+	cacheTTL []byte
+	cacheOperation []byte
 	// correlation id
 	corrID *netstring.Netstring
 	clientinfo *netstring.Netstring
@@ -52,7 +55,6 @@ func NewHeraConnection(conn net.Conn) driver.Conn {
 	return hera
 }
 
-
 // Prepare returns a prepared statement, bound to this connection.
 func (c *heraConnection) Prepare(query string) (driver.Stmt, error) {
 	if logger.GetLogger().V(logger.Debug) {
@@ -100,6 +102,7 @@ func (c *heraConnection) execNs(ns *netstring.Netstring) error {
 		logger.GetLogger().Log(logger.Verbose, c.id, "send command:", ns.Cmd, ", payload:", payload)
 	}
 	_, err := c.conn.Write(ns.Serialized)
+	logger.GetLogger().Log(logger.Verbose, c.id, "after write")
 	return err
 }
 
@@ -171,6 +174,36 @@ func (c *heraConnection) ResetShardKeyPayload() {
 	c.SetShardKeyPayload("")
 }
 
+// implementing the extension HeraConn interface
+func (c *heraConnection) SetCacheKey(payload string) {
+	c.cacheKey = []byte(payload)
+}
+
+// implementing the extension HeraConn interface
+func (c *heraConnection) ResetCacheKey() {
+	c.SetCacheKey("")
+}
+
+// implementing the extension HeraConn interface
+func (c *heraConnection) SetCacheTTL(payload string) {
+	c.cacheTTL = []byte(payload)
+}
+
+// implementing the extension HeraConn interface
+func (c *heraConnection) ResetCacheTTL() {
+	c.SetCacheTTL("")
+}
+
+// implementing the extension HeraConn interface
+func (c *heraConnection) SetCacheOperation(payload string) {
+	c.cacheOperation = []byte(payload)
+}
+
+// implementing the extension HeraConn interface
+func (c *heraConnection) ResetCacheOperation() {
+	c.SetCacheOperation("")
+}
+
 // implementing the extension HeraConn interface
 func (c *heraConnection) SetCalCorrID(corrID string) {
 	c.corrID = netstring.NewNetstringFrom(common.CmdClientCalCorrelationID, []byte(fmt.Sprintf("CorrId=%s", corrID)))
@@ -239,4 +272,36 @@ func (c *heraConnection) SetClientInfoWithPoolStack(poolName string, host string
                 logger.GetLogger().Log(logger.Debug, "Server info:", string(ns.Payload))
         }
 	return nil
-}
+}
+
+func (c *heraConnection) SetClientInfoWithPayload(poolName string, host string, payload string)(error){
+	if len(poolName) <= 0 && len(host) <= 0 && len(payload) <= 0 {
+		return nil
+	}
+
+	pid := os.Getpid()
+	data := fmt.Sprintf("PID: %d, HOST: %s, Poolname: %s, %s, Command: SetClientInfo,", pid, host, poolName, payload)
+	c.clientinfo = netstring.NewNetstringFrom(common.CmdClientInfo, []byte(string(data)))
+	if logger.GetLogger().V(logger.Verbose) {
+		logger.GetLogger().Log(logger.Verbose, "SetClientInfo", c.clientinfo.Serialized)
+	}
+
+	_, err := c.conn.Write(c.clientinfo.Serialized)
+	if err != nil {
+		if logger.GetLogger().V(logger.Warning) {
+			logger.GetLogger().Log(logger.Warning, "Failed to send client info")
+		}
+		return errors.New("Failed custom auth, failed to send client info")
+	}
+	ns, err := c.reader.ReadNext()
+	if err != nil {
+		if logger.GetLogger().V(logger.Warning) {
+			logger.GetLogger().Log(logger.Warning, "Failed to read server info")
+		}
+		return errors.New("Failed to read server info")
+	}
+	if logger.GetLogger().V(logger.Debug) {
+		logger.GetLogger().Log(logger.Debug, "Server info:", string(ns.Payload))
+	}
+	return nil
+}

client/gosqldriver/ext.go

@@ -47,6 +47,26 @@ type HeraConn interface {
 	SetClientInfo(poolname string, host string) error
 
 	SetClientInfoWithPoolStack(poolName string, host string, poolStack string) error
+
+	SetClientInfoWithPayload(poolName string, host string, payload string) error
+
+	// Used to pass the cache key as part of the request.
+	SetCacheKey(payload string)
+
+	// Reset the set set via SetCacheKey
+	ResetCacheKey()
+
+	// Used to set the cache TTL as part of the request.
+	SetCacheTTL(payload string)
+
+	// Reset the set set via SetCacheTTL
+	ResetCacheTTL()
+
+	// Used to set the cache operation. (For ex: GET) as part of the request.
+	SetCacheOperation(payload string)
+
+	// Reset the set set via SetCacheOperation
+	ResetCacheOperation()
 }
 
 // HeraStmt is an API extension for *sql.Stmt