Capture CassandraLWT error and log/bump metrics for it.

common/metrics/defs.go

@@ -1816,6 +1816,7 @@ const (
 	PersistenceErrExecutionAlreadyStartedCounter
 	PersistenceErrDomainAlreadyExistsCounter
 	PersistenceErrBadRequestCounter
+	PersistenceErrDBUnavailableCounter
 	PersistenceSampledCounter
 	PersistenceEmptyResponseCounter
 
@@ -2349,6 +2350,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		PersistenceErrExecutionAlreadyStartedCounter:        {metricName: "persistence_errors_execution_already_started", metricType: Counter},
 		PersistenceErrDomainAlreadyExistsCounter:            {metricName: "persistence_errors_domain_already_exists", metricType: Counter},
 		PersistenceErrBadRequestCounter:                     {metricName: "persistence_errors_bad_request", metricType: Counter},
+		PersistenceErrDBUnavailableCounter:                  {metricName: "persistence_errors_db_unavailable", metricType: Counter},
 		PersistenceSampledCounter:                           {metricName: "persistence_sampled", metricType: Counter},
 		PersistenceEmptyResponseCounter:                     {metricName: "persistence_empty_response", metricType: Counter},
 		CadenceClientRequests:                               {metricName: "cadence_client_requests", metricType: Counter},

common/persistence/dataManagerInterfaces.go

@@ -266,6 +266,11 @@ type (
 		Msg string
 	}
 
+	// DBUnavailableError is returned when the database is unavailable, could be for various reasons.
+	DBUnavailableError struct {
+		Msg string
+	}
+
 	// TransactionSizeLimitError is returned when the transaction size is too large
 	TransactionSizeLimitError struct {
 		Msg string
@@ -1862,6 +1867,10 @@ func (e *TimeoutError) Error() string {
 	return e.Msg
 }
 
+func (e *DBUnavailableError) Error() string {
+	return e.Msg
+}
+
 func (e *TransactionSizeLimitError) Error() string {
 	return e.Msg
 }

common/persistence/nosql/nosqlplugin/cassandra/db.go

@@ -68,3 +68,7 @@ func (db *cdb) IsTimeoutError(err error) bool {
 func (db *cdb) IsThrottlingError(err error) bool {
 	return db.client.IsThrottlingError(err)
 }
+
+func (db *cdb) IsDBUnavailableError(err error) bool {
+	return db.client.IsDBUnavailableError(err)
+}

common/persistence/nosql/nosqlplugin/cassandra/gocql/public/client.go

@@ -22,6 +22,7 @@ package public
 
 import (
 	"context"
+	"strings"
 
 	gogocql "github.com/gocql/gocql"
 
@@ -70,3 +71,16 @@ func (c client) IsThrottlingError(err error) bool {
 	}
 	return false
 }
+
+func (c client) IsDBUnavailableError(err error) bool {
+	if req, ok := err.(gogocql.RequestError); ok {
+		// 0x1000 == UNAVAILABLE
+		if req.Code() != 0x1000 {
+			return false
+		}
+		if strings.Contains(req.Message(), "Cannot perform LWT operation") {
+			return true
+		}
+	}
+	return false
+}

common/persistence/nosql/nosqlplugin/dynamodb/db.go

@@ -69,6 +69,10 @@ func (db *ddb) IsThrottlingError(err error) bool {
 	panic("TODO")
 }
 
+func (db *ddb) IsDBUnavailableError(err error) bool {
+	panic("TODO")
+}
+
 func (db *ddb) IsConditionFailedError(err error) bool {
 	return err == errConditionFailed
 }

common/persistence/nosql/nosqlplugin/interfaces.go

@@ -76,6 +76,7 @@ type (
 		IsTimeoutError(error) bool
 		IsNotFoundError(error) bool
 		IsThrottlingError(error) bool
+		IsDBUnavailableError(error) bool
 	}
 
 	/**

common/persistence/nosql/nosqlplugin/mongodb/error.go

@@ -35,3 +35,7 @@ func (db *mdb) IsTimeoutError(err error) bool {
 func (db *mdb) IsThrottlingError(err error) bool {
 	return false
 }
+
+func (db *mdb) IsDBUnavailableError(err error) bool {
+	return false
+}

common/persistence/nosql/utils.go

@@ -49,6 +49,12 @@ func convertCommonErrors(
 		}
 	}
 
+	if errChecker.IsDBUnavailableError(err) {
+		return &p.DBUnavailableError{
+			Msg: fmt.Sprintf("#{operation} operation failed. Error: #{err}"),
+		}
+	}
+
 	return &types.InternalServiceError{
 		Message: fmt.Sprintf("%v operation failed. Error: %v", operation, err),
 	}

common/persistence/persistenceMetricClients.go

@@ -248,6 +248,10 @@ func (p *persistenceMetricsClientBase) updateErrorMetric(scope int, err error) {
 	case *types.ServiceBusyError:
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrBusyCounter)
 		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)
+	case *DBUnavailableError:
+		p.metricClient.IncCounter(scope, metrics.PersistenceErrDBUnavailableCounter)
+		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)
+		p.logger.Error("DBUnavailable Error:", tag.Error(err), tag.MetricScope(scope))
 	default:
 		p.logger.Error("Operation failed with internal error.", tag.Error(err), tag.MetricScope(scope))
 		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)