Ignore only Insights RPC

Author: lukasz-antoniak

proxy/pkg/zdmproxy/cqlparser.go

@@ -23,6 +23,10 @@ const (
 	forwardToAsyncOnly = forwardDecision("async") // for "synchronous" requests that should be sent to the async connector (handshake requests)
 )
 
+const (
+	insightsRpcName = "InsightsRpc"
+)
+
 type interceptedQueryType string
 
 const (
@@ -215,10 +219,12 @@ func getRequestInfoFromQueryInfo(
 	} else if queryInfo.getStatementType() == statementTypeUse {
 		sendAlsoToAsync = true
 	} else if queryInfo.getStatementType() == statementTypeCall {
-		// RPC client, typically used by "insights reporter" for DSE
-		forwardDecision = forwardToNone
-		trackMetrics = false
-		sendAlsoToAsync = false
+		if queryInfo.getCallRpcName() == insightsRpcName {
+			// ignore Insights client calls
+			forwardDecision = forwardToNone
+			trackMetrics = false
+			sendAlsoToAsync = false
+		}
 	} else {
 		sendAlsoToAsync = false
 	}

proxy/pkg/zdmproxy/cqlparser_test.go

@@ -95,6 +95,7 @@ func TestInspectFrame(t *testing.T) {
 		{"OpCodeQuery UNKNOWN", args{mockQueryFrame(t, "UNKNOWN"), []*term{}, primaryClusterOrigin, forwardSystemQueriesToOrigin, forwardAuthToOrigin}, NewGenericRequestInfo(forwardToBoth, false, true)},
 		{"OpCodeQuery CALL InsightsRpc.reportInsight(?)", args{mockQueryFrame(t, "CALL InsightsRpc.reportInsight(?)"), []*term{}, primaryClusterOrigin, forwardSystemQueriesToOrigin, forwardAuthToOrigin}, NewGenericRequestInfo(forwardToNone, false, false)},
 		{"OpCodeQuery CALL InsightsRpc.reportInsight('a', 1, -2.3, true, '2020-01-01')", args{mockQueryFrame(t, "CALL InsightsRpc.reportInsight('a', 1, -2.3, true, '2020-01-01')"), []*term{}, primaryClusterOrigin, forwardSystemQueriesToOrigin, forwardAuthToOrigin}, NewGenericRequestInfo(forwardToNone, false, false)},
+		{"OpCodeQuery CALL DseGraphRpc.getSchemaBlob(?)", args{mockQueryFrame(t, "CALL DseGraphRpc.getSchemaBlob(?)"), []*term{}, primaryClusterOrigin, forwardSystemQueriesToOrigin, forwardAuthToOrigin}, NewGenericRequestInfo(forwardToBoth, false, true)},
 
 		// PREPARE
 		{"OpCodePrepare SELECT", args{mockPrepareFrame(t, "SELECT blah FROM ks1.t1"), []*term{}, primaryClusterOrigin, forwardSystemQueriesToOrigin, forwardAuthToOrigin}, NewPrepareRequestInfo(NewGenericRequestInfo(forwardToOrigin, true, true), []*term{}, false, "SELECT blah FROM ks1.t1", "")},

proxy/pkg/zdmproxy/queryinspector.go

@@ -52,6 +52,7 @@ type QueryInfo interface {
 	getStatementType() statementType
 	getKeyspaceName() string
 	getTableName() string
+	getCallRpcName() string
 
 	// Returns the "current" keyspace when this request was parsed. This could have been set by a "USE" request beforehand
 	// or by using the keyspace query/prepare flag in v5 or DseV2.
@@ -327,6 +328,7 @@ type cqlListener struct {
 	timeUuidGenerator TimeUuidGenerator
 
 	requestKeyspace string
+	callRpcName     string
 }
 
 func (l *cqlListener) getQuery() string {
@@ -345,6 +347,10 @@ func (l *cqlListener) getTableName() string {
 	return l.tableName
 }
 
+func (l *cqlListener) getCallRpcName() string {
+	return l.callRpcName
+}
+
 func (l *cqlListener) getRequestKeyspace() string {
 	return l.requestKeyspace
 }
@@ -401,6 +407,19 @@ func (l *cqlListener) EnterCqlStatement(ctx *parser.CqlStatementContext) {
 	}
 }
 
+func (l *cqlListener) EnterCallStatement(ctx *parser.CallStatementContext) {
+	if ctx.GetChildCount() < 2 {
+		return
+	}
+
+	// RPC name will be present as second child: CALL <rpc_name>.<rpc_method>(<params>)
+	token := ctx.GetChild(1)
+	switch token.(type) {
+	case *parser.IdentifierContext:
+		l.callRpcName = token.(*parser.IdentifierContext).GetText()
+	}
+}
+
 func (l *cqlListener) ExitSelectStatement(ctx *parser.SelectStatementContext) {
 	if !isSystemKeyspace(l.getApplicableKeyspace()) {
 		return