[#16] .Update methods for parsing the response headers from cosmos

cosmos.go

@@ -139,3 +139,110 @@ func (c *Cosmos) String() string {
 func (c *Cosmos) IsHealthy() error {
 	return c.pool.Ping()
 }
+
+func parseAttributeMap(attributes map[string]interface{}) (responseInformation, error) {
+
+	responseInfo := responseInformation{}
+	if valueStr, ok := attributes[string(headerStatusCode)]; ok {
+
+		value, err := cast.ToInt16E(valueStr)
+		if err != nil {
+			return responseInfo, errors.Wrapf(err, "Failed parsing '%s'", headerStatusCode)
+		}
+		statusCode := int(value)
+		responseInfo.statusCode = statusCode
+		responseInfo.statusDescription = statusCodeToDescription(statusCode)
+	}
+
+	if valueStr, ok := attributes[string(headerSubStatusCode)]; ok {
+		responseInfo.subStatusCode = int(cast.ToInt16(valueStr))
+	}
+
+	if valueStr, ok := attributes[string(headerRequestCharge)]; ok {
+		responseInfo.requestCharge = cast.ToFloat32(valueStr)
+	}
+
+	if valueStr, ok := attributes[string(headerRequestChargeTotal)]; ok {
+		responseInfo.requestChargeTotal = cast.ToFloat32(valueStr)
+	}
+
+	if valueStr, ok := attributes[string(headerServerTimeMS)]; ok {
+		responseInfo.serverTime = time.Microsecond * time.Duration(1000*cast.ToFloat32(valueStr))
+	}
+
+	if valueStr, ok := attributes[string(headerServerTimeMSTotal)]; ok {
+		responseInfo.serverTimeTotal = time.Microsecond * time.Duration(1000*cast.ToFloat32(valueStr))
+	}
+
+	if valueStr, ok := attributes[string(headerActivityID)]; ok {
+		responseInfo.activityID = cast.ToString(valueStr)
+	}
+
+	if valueStr, ok := attributes[string(headerRetryAfterMS)]; ok {
+		retryAfter, err := time.Parse("15:04:05.999999999", cast.ToString(valueStr))
+		zeroTime, _ := time.Parse("15:04:05.999999999", "00:00:00.000")
+		responseInfo.retryAfter = retryAfter.Sub(zeroTime)
+		if err != nil {
+			responseInfo.retryAfter = 0
+		}
+	}
+
+	if valueStr, ok := attributes[string(headerSource)]; ok {
+		responseInfo.source = cast.ToString(valueStr)
+	}
+
+	return responseInfo, nil
+}
+
+func statusCodeToDescription(code int) string {
+	desc, ok := statusCodeDescription[code]
+	if !ok {
+		return fmt.Sprintf("Status code %d is unknown", code)
+	}
+	return desc
+}
+
+type responseInformation struct {
+	statusCode         int
+	subStatusCode      int
+	statusDescription  string
+	requestCharge      float32
+	requestChargeTotal float32
+	serverTime         time.Duration
+	serverTimeTotal    time.Duration
+	activityID         string
+	retryAfter         time.Duration
+	source             string
+}
+
+// statusCodeDescription provides the description for status codes taken from https://docs.microsoft.com/en-us/azure/cosmos-db/gremlin-headers#status-codes
+var statusCodeDescription = map[int]string{
+	401:  "Error message 'Unauthorized: Invalid credentials provided' is returned when authentication password doesn't match Cosmos DB account key. Navigate to your Cosmos DB Gremlin account in the Azure portal and confirm that the key is correct.",
+	404:  "Concurrent operations that attempt to delete and update the same edge or vertex simultaneously. Error message 'Owner resource does not exist' indicates that specified database or collection is incorrect in connection parameters in /dbs/<database name>/colls/<collection or graph name> format.",
+	408:  "'Server timeout' indicates that traversal took more than 30 seconds and was canceled by the server. Optimize your traversals to run quickly by filtering vertices or edges on every hop of traversal to narrow down search scope.",
+	409:  "'Conflicting request to resource has been attempted. Retry to avoid conflicts.' This usually happens when vertex or an edge with an identifier already exists in the graph.",
+	412:  "Status code is complemented with error message 'PreconditionFailedException': One of the specified pre-condition is not met. This error is indicative of an optimistic concurrency control violation between reading an edge or vertex and writing it back to the store after modification. Most common situations when this error occurs is property modification, for example g.V('identifier').property('name','value'). Gremlin engine would read the vertex, modify it, and write it back. If there is another traversal running in parallel trying to write the same vertex or an edge, one of them will receive this error. Application should submit traversal to the server again.",
+	429:  "Request was throttled and should be retried after value in x-ms-retry-after-ms",
+	500:  "Error message that contains 'NotFoundException: Entity with the specified id does not exist in the system.' indicates that a database and/or collection was re-created with the same name. This error will disappear within 5 minutes as change propagates and invalidates caches in different Cosmos DB components. To avoid this issue, use unique database and collection names every time.",
+	1000: "This status code is returned when server successfully parsed a message but wasn't able to execute. It usually indicates a problem with the query.",
+	1001: "This code is returned when server completes traversal execution but fails to serialize response back to the client. This error can happen when traversal generates complex result, that is too large or does not conform to TinkerPop protocol specification. Application should simplify the traversal when it encounters this error.",
+	1003: "'Query exceeded memory limit. Bytes Consumed: XXX, Max: YYY' is returned when traversal exceeds allowed memory limit. Memory limit is 2 GB per traversal.",
+	1004: "This status code indicates malformed graph request. Request can be malformed when it fails deserialization, non-value type is being deserialized as value type or unsupported gremlin operation requested. Application should not retry the request because it will not be successful.",
+	1007: "Usually this status code is returned with error message 'Could not process request. Underlying connection has been closed.'. This situation can happen if client driver attempts to use a connection that is being closed by the server. Application should retry the traversal on a different connection.",
+	1008: "Cosmos DB Gremlin server can terminate connections to rebalance traffic in the cluster. Client drivers should handle this situation and use only live connections to send requests to the server. Occasionally client drivers may not detect that connection was closed. When application encounters an error, 'Connection is too busy. Please retry after sometime or open more connections.' it should retry traversal on a different connection.",
+}
+
+// Responseheaders for CosmosDB, taken from: https://docs.microsoft.com/en-us/azure/cosmos-db/gremlin-headers#headers
+type cosmosDBResponseHeader string
+
+const (
+	headerRequestCharge      cosmosDBResponseHeader = "x-ms-request-charge"       // double
+	headerRequestChargeTotal cosmosDBResponseHeader = "x-ms-total-request-charge" // double
+	headerServerTimeMS       cosmosDBResponseHeader = "x-ms-server-time-ms"       // double
+	headerServerTimeMSTotal  cosmosDBResponseHeader = "x-ms-total-server-time-ms" // double
+	headerStatusCode         cosmosDBResponseHeader = "x-ms-status-code"          // long
+	headerSubStatusCode      cosmosDBResponseHeader = "x-ms-substatus-code"       // long
+	headerRetryAfterMS       cosmosDBResponseHeader = "x-ms-retry-after-ms"       // string
+	headerActivityID         cosmosDBResponseHeader = "x-ms-activity-id"          // string
+	headerSource             cosmosDBResponseHeader = "x-ms-source"               // string
+)

cosmos_test.go

@@ -76,3 +76,93 @@ func TestIsHealthy(t *testing.T) {
 	assert.NoError(t, healthyWhenConnected)
 	assert.Error(t, healthyWhenNotConnected)
 }
+
+func TestParseAttributeMap(t *testing.T) {
+	// GIVEN
+	attributeMap := map[string]interface{}{
+		"x-ms-status-code":          429,
+		"x-ms-substatus-code":       3200,
+		"x-ms-request-charge":       1234.56,
+		"x-ms-total-request-charge": 78910.11,
+		"x-ms-server-time-ms":       11.22,
+		"x-ms-total-server-time-ms": 333.444,
+		"x-ms-activity-id":          "fdd08592-abcd-efgh-ijkl-97d35c2dda52",
+		"x-ms-retry-after-ms":       "00:00:02.345",
+		"x-ms-source":               "Microsoft.Azure.Documents.Client",
+	}
+
+	// WHEN
+	responseInfo, err := parseAttributeMap(attributeMap)
+
+	// THEN
+	require.NoError(t, err)
+	assert.Equal(t, 429, responseInfo.statusCode)
+	assert.NotEmpty(t, responseInfo.statusDescription)
+	assert.Equal(t, 3200, responseInfo.subStatusCode)
+	assert.Equal(t, float32(1234.56), responseInfo.requestCharge)
+	assert.Equal(t, float32(78910.11), responseInfo.requestChargeTotal)
+	assert.Equal(t, time.Microsecond*11220, responseInfo.serverTime)
+	assert.Equal(t, time.Microsecond*333444, responseInfo.serverTimeTotal)
+	assert.Equal(t, "fdd08592-abcd-efgh-ijkl-97d35c2dda52", responseInfo.activityID)
+	assert.Equal(t, time.Millisecond*2345, responseInfo.retryAfter)
+	assert.Equal(t, "Microsoft.Azure.Documents.Client", responseInfo.source)
+}
+
+func TestParseAttributeMapFail(t *testing.T) {
+	// GIVEN
+	attributeMap := map[string]interface{}{
+		"x-ms-status-code": "invalid",
+	}
+
+	// WHEN
+	responseInfo, err := parseAttributeMap(attributeMap)
+
+	// THEN
+	require.Error(t, err)
+
+	// GIVEN
+	attributeMap = map[string]interface{}{
+		"x-ms-status-code":          429,
+		"x-ms-substatus-code":       "invalid",
+		"x-ms-request-charge":       "invalid",
+		"x-ms-total-request-charge": "invalid",
+		"x-ms-server-time-ms":       "invalid",
+		"x-ms-total-server-time-ms": "invalid",
+		"x-ms-retry-after-ms":       "invalid",
+	}
+
+	// WHEN
+	responseInfo, err = parseAttributeMap(attributeMap)
+
+	// THEN
+	require.NoError(t, err)
+	assert.Equal(t, 429, responseInfo.statusCode)
+	assert.NotEmpty(t, responseInfo.statusDescription)
+	assert.Equal(t, 0, responseInfo.subStatusCode)
+	assert.Equal(t, float32(0), responseInfo.requestCharge)
+	assert.Equal(t, float32(0), responseInfo.requestChargeTotal)
+	assert.Equal(t, time.Microsecond*0, responseInfo.serverTime)
+	assert.Equal(t, time.Microsecond*0, responseInfo.serverTimeTotal)
+	assert.Equal(t, time.Millisecond*0, responseInfo.retryAfter)
+}
+
+func TestStatusCodeToDescription(t *testing.T) {
+	// GIVEN
+	code := 429
+
+	// WHEN
+	desc := statusCodeToDescription(code)
+
+	// THEN
+	assert.Contains(t, desc, "throttled")
+	assert.NotContains(t, desc, "unknown")
+
+	// GIVEN -- not found
+	code = 12345
+
+	// WHEN
+	desc = statusCodeToDescription(code)
+
+	// THEN
+	assert.Contains(t, desc, "unknown")
+}

examples/cosmos/main.go

@@ -12,14 +12,6 @@ import (
 	gremcos "github.com/supplyon/gremcos"
 )
 
-var panicOnErrorOnChannel = func(errs chan error) {
-	err := <-errs
-	if err == nil {
-		return // ignore if the channel was closed
-	}
-	log.Fatalf("Lost connection to the database: %s", err)
-}
-
 func main() {
 
 	host := os.Getenv("CDB_HOST")

go.mod

@@ -8,5 +8,6 @@ require (
 	github.com/gorilla/websocket v1.4.1
 	github.com/pkg/errors v0.9.1
 	github.com/rs/zerolog v1.18.0
+	github.com/spf13/cast v1.3.1
 	github.com/stretchr/testify v1.4.0
 )

go.sum

@@ -1,6 +1,8 @@
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
 github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/golang/mock v1.4.0 h1:Rd1kQnQu0Hq3qvJppYSG0HtP+f5LPPUiDswTLiEegLg=
@@ -15,7 +17,10 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/rs/zerolog v1.18.0 h1:CbAm3kP2Tptby1i9sYy2MGRg0uxIN9cyDb59Ys7W8z8=
 github.com/rs/zerolog v1.18.0/go.mod h1:9nvC1axdVrAHcu/s9taAVfBuIdTZLVQmKQyvrUjF5+I=
+github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
+github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=