fixes #407: Move the internal procedures calls to direct API calls (#415)

conker84 · web-flow · commit 8e67bceef6b8 · 2021-06-07T19:55:10.000+02:00
diff --git a/common/src/main/kotlin/streams/config/StreamsConfig.kt b/common/src/main/kotlin/streams/config/StreamsConfig.kt
@@ -15,6 +15,7 @@ import streams.extensions.databaseManagementService
 import streams.extensions.getDefaultDbName
 import streams.extensions.isAvailable
 import streams.utils.Neo4jUtils
+import streams.utils.ProcedureUtils
 import streams.utils.StreamsUtils
 import java.io.File
 import java.util.concurrent.ConcurrentHashMap
@@ -117,7 +118,7 @@ class StreamsConfig(private val log: Log, private val dbms: DatabaseManagementSe
                     log.warn("${getInstanceWaitTimeout()} ms have passed and the instance is not online, the Streams plugin will not started")
                     return@runBlocking
                 }
-                if (Neo4jUtils.isCluster(dbms, log)) {
+                if (ProcedureUtils.isCluster(dbms)) {
                     log.info("We're in cluster instance waiting for the ${StreamsUtils.LEADER}s to be elected in each database")
                     // in case is a cluster we wait for the correct cluster formation => LEADER elected
                     Neo4jUtils.waitForTheLeaders(dbms, log) { configStart() }
diff --git a/common/src/main/kotlin/streams/utils/Neo4jUtils.kt b/common/src/main/kotlin/streams/utils/Neo4jUtils.kt
@@ -8,7 +8,6 @@ import org.neo4j.dbms.api.DatabaseManagementService
 import org.neo4j.graphdb.QueryExecutionException
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
-import org.neo4j.logging.internal.LogService
 import streams.extensions.execute
 import java.lang.reflect.InvocationTargetException
 import kotlin.streams.toList
@@ -26,7 +25,7 @@ object Neo4jUtils {
                 return false
             }
 
-            return availableAction() && getMemberRole(db).equals(StreamsUtils.LEADER, ignoreCase = true)
+            return availableAction() && ProcedureUtils.clusterMemberRole(db).equals(StreamsUtils.LEADER, ignoreCase = true)
         } catch (e: QueryExecutionException) {
             if (e.statusCode.equals("Neo.ClientError.Procedure.ProcedureNotFound", ignoreCase = true)) {
                 return availableAction()
@@ -44,36 +43,7 @@ object Neo4jUtils {
         false
     }
 
-    fun getLogService(db: GraphDatabaseAPI): LogService = db.dependencyResolver
-            .resolveDependency(LogService::class.java)
-
-    fun isCluster(db: GraphDatabaseAPI, log: Log? = null): Boolean {
-        try {
-            return db.execute("CALL dbms.cluster.overview()") {
-                if (it.hasNext()) {
-                    if (log?.isDebugEnabled == true) {
-                        log?.debug(it.resultAsString())
-                    }
-                }
-                true
-            }
-        } catch (e: QueryExecutionException) {
-            if (e.statusCode.equals("Neo.ClientError.Procedure.ProcedureNotFound", ignoreCase = true)) {
-                return false
-            }
-            throw e
-        }
-    }
-
-    fun isCluster(dbms: DatabaseManagementService, log: Log? = null): Boolean = dbms.listDatabases()
-        .firstOrNull { it != StreamsUtils.SYSTEM_DATABASE_NAME }
-        ?.let { dbms.database(it) as GraphDatabaseAPI }
-        ?.let { isCluster(it, log) } ?: false
-
-    private fun getMemberRole(db: GraphDatabaseAPI) = db.execute("CALL dbms.cluster.role(\$database)",
-            mapOf("database" to db.databaseName())) { it.columnAs<String>("role").next() }
-
-    fun clusterHasLeader(db: GraphDatabaseAPI): Boolean = try {
+    private fun clusterHasLeader(db: GraphDatabaseAPI): Boolean = try {
         db.execute("""
                 |CALL dbms.cluster.overview() YIELD databases
                 |RETURN databases[${'$'}database] AS role
@@ -98,34 +68,6 @@ object Neo4jUtils {
         null
     }
 
-    fun executeInLeader(db: GraphDatabaseAPI,
-                        log: Log,
-                        timeout: Long = 120000,
-                        availableAction: () -> Boolean,
-                        action: () -> Unit) {
-        GlobalScope.launch(Dispatchers.IO) {
-            val start = System.currentTimeMillis()
-            val delay: Long = 2000
-            while (!clusterHasLeader(db) && System.currentTimeMillis() - start < timeout) {
-                log.info("${StreamsUtils.LEADER} not found, new check comes in $delay milliseconds...")
-                delay(delay)
-            }
-            executeInWriteableInstance(db, availableAction, action)
-        }
-    }
-
-    fun waitForTheLeader(db: GraphDatabaseAPI, log: Log, timeout: Long = 120000, action: () -> Unit) {
-        GlobalScope.launch(Dispatchers.IO) {
-            val start = System.currentTimeMillis()
-            val delay: Long = 2000
-            while (!clusterHasLeader(db) && System.currentTimeMillis() - start < timeout) {
-                log.info("${StreamsUtils.LEADER} not found, new check comes in $delay milliseconds...")
-                delay(delay)
-            }
-            action()
-        }
-    }
-
     fun isClusterCorrectlyFormed(dbms: DatabaseManagementService) = dbms.listDatabases()
         .filterNot { it == StreamsUtils.SYSTEM_DATABASE_NAME }
         .map { dbms.database(it) as GraphDatabaseAPI }
diff --git a/common/src/main/kotlin/streams/utils/ProcedureUtils.kt b/common/src/main/kotlin/streams/utils/ProcedureUtils.kt
@@ -0,0 +1,65 @@
+package streams.utils
+
+import org.neo4j.dbms.api.DatabaseManagementService
+import org.neo4j.exceptions.UnsatisfiedDependencyException
+import org.neo4j.kernel.impl.factory.DbmsInfo
+import org.neo4j.kernel.internal.GraphDatabaseAPI
+import org.neo4j.logging.internal.LogService
+import streams.extensions.execute
+import java.lang.invoke.MethodHandles
+import java.lang.invoke.MethodType
+
+
+object ProcedureUtils {
+    @JvmStatic
+    private val coreMetadata: Class<*>? = try {
+        Class.forName("com.neo4j.causalclustering.core.consensus.CoreMetaData")
+    } catch (e: ClassNotFoundException) {
+        null
+    }
+
+    @JvmStatic
+    private val isLeaderMethodHandle = coreMetadata?.let {
+        val lookup = MethodHandles.lookup()
+        lookup.findVirtual(it, "isLeader", MethodType.methodType(Boolean::class.java))
+            .asType(MethodType.methodType(Boolean::class.java, Any::class.java))
+    }
+
+    fun clusterMemberRole(db: GraphDatabaseAPI): String {
+        val fallback: (Exception?) -> String = { e: Exception? ->
+            val userLog = db.dependencyResolver
+                .resolveDependency(LogService::class.java)
+                .getUserLog(ProcedureUtils::class.java)
+            e?.let { userLog.warn("Cannot call the APIs, trying with the cypher query", e) }
+                ?: userLog.warn("Cannot call the APIs, trying with the cypher query")
+            db.execute("CALL dbms.cluster.role(\$database)",
+                mapOf("database" to db.databaseName())
+            ) { it.columnAs<String>("role").next() }
+        }
+        val execute = {
+            coreMetadata?.let {
+                try {
+                    val raftMachine: Any = db.dependencyResolver.resolveDependency(coreMetadata)
+                    val isLeader = isLeaderMethodHandle!!.invokeExact(raftMachine) as Boolean
+                    if (isLeader) "LEADER" else "FOLLOWER"
+                } catch (e: UnsatisfiedDependencyException) {
+                    "LEADER"
+                }
+            } ?: "LEADER"
+        }
+        return executeOrFallback(execute, fallback)
+    }
+
+    fun isCluster(db: GraphDatabaseAPI): Boolean = db.dbmsInfo() == DbmsInfo.CORE || db.dbmsInfo() == DbmsInfo.READ_REPLICA
+
+    fun isCluster(dbms: DatabaseManagementService): Boolean = dbms.listDatabases()
+        .firstOrNull { it != StreamsUtils.SYSTEM_DATABASE_NAME }
+        ?.let { dbms.database(it) as GraphDatabaseAPI }
+        ?.let { isCluster(it) } ?: false
+
+    private fun <T> executeOrFallback(execute: () -> T, fallback: (Exception?) -> T): T = try {
+        execute()
+    } catch (e: Exception) {
+        fallback(e)
+    }
+}
diff --git a/common/src/test/kotlin/streams/utils/Neo4jUtilsTest.kt b/common/src/test/kotlin/streams/utils/Neo4jUtilsTest.kt
@@ -19,12 +19,6 @@ class Neo4jUtilsTest {
         assertTrue { isWriteableInstance }
     }
 
-    @Test
-    fun shouldCheckIfIsACluster() {
-        val isEnterprise = Neo4jUtils.isCluster(db)
-        assertFalse { isEnterprise }
-    }
-
     @Test
     fun `should not have APOC`() {
         assertFalse { Neo4jUtils.hasApoc(db) }
diff --git a/common/src/test/kotlin/streams/utils/ProcedureUtilsTest.kt b/common/src/test/kotlin/streams/utils/ProcedureUtilsTest.kt
@@ -0,0 +1,21 @@
+package streams.utils
+
+import org.junit.ClassRule
+import org.junit.Test
+import org.neo4j.test.rule.ImpermanentDbmsRule
+import kotlin.test.assertFalse
+
+class ProcedureUtilsTest {
+
+    companion object {
+        @ClassRule @JvmField
+        val db = ImpermanentDbmsRule()
+    }
+
+    @Test
+    fun shouldCheckIfIsACluster() {
+        val isCluster = ProcedureUtils.isCluster(db)
+        assertFalse { isCluster }
+    }
+
+}
diff --git a/consumer/src/main/kotlin/streams/StreamsSinkConfigurationListener.kt b/consumer/src/main/kotlin/streams/StreamsSinkConfigurationListener.kt
@@ -19,6 +19,7 @@ import streams.procedures.StreamsSinkProcedures
 import streams.utils.ConsumerUtils
 import streams.utils.KafkaValidationUtils
 import streams.utils.Neo4jUtils
+import streams.utils.ProcedureUtils
 import streams.utils.StreamsUtils
 
 class StreamsSinkConfigurationListener(private val db: GraphDatabaseAPI,
@@ -119,7 +120,7 @@ class StreamsSinkConfigurationListener(private val db: GraphDatabaseAPI,
         try {
             if (streamsSinkConfiguration.enabled) {
                 log.info("[Sink] The Streams Sink module is starting")
-                if (Neo4jUtils.isCluster(db, log)) {
+                if (ProcedureUtils.isCluster(db)) {
                     initSinkModule(streamsSinkConfiguration)
                 } else {
                     runInASingleInstance(streamsSinkConfiguration)
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt b/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt
@@ -27,6 +27,7 @@ import streams.extensions.isDefaultDb
 import streams.utils.ConsumerUtils
 import streams.utils.KafkaValidationUtils.getInvalidTopicsError
 import streams.utils.Neo4jUtils
+import streams.utils.ProcedureUtils
 import streams.utils.StreamsUtils
 
 
@@ -72,7 +73,7 @@ class KafkaEventSink(private val config: Map<String, String>,
     }
 
     override fun start() = runBlocking { // TODO move to the abstract class
-        if (streamsConfig.clusterOnly && !Neo4jUtils.isCluster(db)) {
+        if (streamsConfig.clusterOnly && !ProcedureUtils.isCluster(db)) {
             if (log.isDebugEnabled) {
                 log.info("""
                         |Cannot init the Kafka Sink module as is forced to work only in a cluster env, 
diff --git a/doc/package-lock.json b/doc/package-lock.json
diff --git a/producer/src/main/kotlin/streams/StreamsConstraintsService.kt b/producer/src/main/kotlin/streams/StreamsConstraintsService.kt
