Throughput degradation (Gremlin Server CPU usage at 130%)

[rogaha]

I've been using Janusgraph in production for fraud prevention scenarios, but recently we have been experiencing an ingestion throughput degradation. Our backend is Cassandra/Elasticsearch and below are the current ingestion statistics:

35820206 [metrics-logger-reporter-thread-1] INFO org.apache.tinkerpop.gremlin.server.Settings$Slf4jReporterMetrics - type=TIMER, name=org.apache.tinkerpop.gremlin.server.GremlinServer.op.eval, count=83902, min=346.476045, max=23306.430972, mean=5669.135124586575, stddev=4673.183770559961, median=4345.621524499999, p75=8064.485307999999, p95=15156.586356099986, p98=17926.15823546, p99=18811.370566920003, p999=23280.965525466003, mean_rate=2.343088469134227, m1=1.7210005889641737, m5=1.786056502065271, m15=1.9371755852213393, rate_unit=events/second, duration_unit=milliseconds

In this case, we are ingesting lots of payments in an up-insert fashion. Our Janusgraph instance has 32GB of memory and 8 cores with the following Gremlin/JVM settings:

threadPoolWorker: 16
gremlinPool: 8

----------------------------------

storage.backend=cql
storage.hostname=cassandra-prod.cassandra.svc.cluster.local
storage.username=cassandra
storage.password=cassandra
storage.cql.keyspace=aegis
storage.cql.read-consistency-level=QUORUM
storage.cql.write-consistency-level=QUORUM
storage.buffer-size=4096
log.tx.key-consistent=false

cache.db-cache = true
cache.db-cache-clean-wait = 20
cache.db-cache-time = 1500000
cache.db-cache-size = 0.40
cache.tx-dirty-size = 2048
cache.tx-cache-size = 20000

graph.replace-instance-if-exists=true

index.search.backend=elasticsearch
index.search.hostname=elasticsearch-master-0.elasticsearch-master-headless.janusgraph.svc.cluster.local
index.search.index-name=aegis
index.search.elasticsearch.client-only=true
query.force-index=true
query.batch=true

ids.block-size=1000000
ids.authority.wait-time=1000

----------------------------------

-XX:+UseStringDeduplication 
-XX:ConcGCThreads=4 
-XX:ParallelGCThreads=4 
-XX:+UseG1GC 
-XX:G1RSetUpdatingPauseTimePercent=5
-XX:MaxGCPauseMillis=500 
-Xms16g -Xmx16g 
-javaagent:/opt/janusgraph/prometheus/jmx_prometheus_javaagent-0.14.0.jar=8090:/opt/janusgraph/prometheus/config.yml -javaagent:/opt/janusgraph/lib/jamm-0.3.0.jar 
-Dgremlin.io.kryoShimService=org.janusgraph.hadoop.serialize.JanusGraphKryoShimService

I've noticed that when I pause the ingestion (payments worker) for a while, the ingestion throughput goes up, temporarily, after I restart the worker and the amount of time with high throughput is directly correlated to the time the worker is paused/turned-off. My first hypothesis is that this is related to the ID allocation process (ids.block-size/ids.authority.wait-time), but I'm not 100% sure.

Cassandra throughput after a few worker restarts with different timeframes

Does anyone that understands Janusgraph implementation can help me clarify what's going on?

See bellow some stats from the Janusgraph instance.

Stats while the ingestion is at high throughput

Stats while the ingestion is at low throughput (few minutes after)

[rogaha]

mystery solved: we had a unique index defined to guarantee the unicity of the node IDs. Removing this index improved the throughput by ~10x.

[rogaha]

thanks for the pointers @porunov. I appreciate it.

[porunov]

@rogaha you profiling shows that you are using LuceneIndex which is part of janusgraph-lucene and not janusgraph-es.
You were probably using a wrong JanusGraph installation during profiling. Could you please profile your usage again using master branch and CQL + ES modules? Would be great to understand where your queries spend most of the time. When we understand that, we could check the logic behind it

[rogaha]

I'm using Lucene now... Our current deployment has only one node.

[rogaha]

Is the profiling above enough?

[mad]

@li-boxuan regarding the usage pattern, I'm doing upserts to avoid node duplication. I use the following to insert nodes & edges:

g.V().has('node_id', '%[2]s').
		fold().coalesce(unfold(), addV('%[1]s').property('node_id', '%[2]s'))
-----
g.V().has('node_id', '%[1]s').as('source').
		      V().has('node_id', '%[2]s').
				coalesce(__.inE('%[3]s').where(outV().as('source')),
				addE('%[3]s').property("timestamp", System.currentTimeMillis()).from('source'))

Is it a common pattern?

@rogaha

This query is not optimal because edge exists does not use VCI

graph=JanusGraphFactory.open("inmemory")
g=graph.traversal()
graph.addVertex()
graph.addVertex()

g.V().limit(1).as('source').V().limit(1).coalesce(__.inE('%[3]s').where(outV().as('source')),addE('%[3]s').property("timestamp", System.currentTimeMillis()).from('source')).profile()

Profile is

==>Traversal Metrics
Step                                                               Count  Traversers       Time (ms)    % Dur
=============================================================================================================
JanusGraphStep(vertex,[])@[source]                                     1           1           0.297     7.07
  constructGraphCentricQuery                                                                   0.018
  constructGraphCentricQuery                                                                   0.012
  GraphCentricQuery                                                                            4.051
    \_condition=()
    \_orders=[]
    \_limit=1
    \_isFitted=false
    \_isOrdered=true
    \_query=[]
    scan                                                                                       0.047
    \_query=[]
    \_fullscan=true
    \_condition=VERTEX
JanusGraphStep(vertex,[])                                              1           1           0.208     4.95
  constructGraphCentricQuery                                                                   0.010
  constructGraphCentricQuery                                                                   0.008
  GraphCentricQuery                                                                            3.792
    \_condition=()
    \_orders=[]
    \_limit=1
    \_isFitted=false
    \_isOrdered=true
    \_query=[]
    scan                                                                                       0.029
    \_query=[]
    \_fullscan=true
    \_condition=VERTEX
CoalesceStep([[JanusGraphVertexStep(IN,[%[3]s],...                     1           1           3.707    87.98
  JanusGraphVertexStep(IN,[%[3]s],edge)                                1           1           3.456
    \_condition=type[%[3]s]
    \_orders=[]
    \_isFitted=true
    \_isOrdered=true
    \_query=%[3]s:SliceQuery[0x70A1,0x70A2)
    \_vertices=1
    optimization                                                                               3.236
  WhereTraversalStep([WhereStartStep, EdgeVerte...                     1           1           0.160
    WhereStartStep                                                     1           1           0.028
    EdgeVertexStep(OUT)                                                1           1           0.034
    WhereEndStep(source)                                                                       0.019
  AddEdgeStep({~from=[[SelectOneStep(last,sourc...                                             0.000
    SelectOneStep(last,source)                                                                 0.000
                                            >TOTAL                     -           -           4.214        -

In coalesce part JanusGraphVertexStep will retrieve all edges of specified type and filter out in memory (WhereTraversalStep)

CoalesceStep([[JanusGraphVertexStep(IN,[%[3]s],...                     1           1           3.707    87.98
  JanusGraphVertexStep(IN,[%[3]s],edge)                                1           1           3.456
    \_condition=type[%[3]s]
    \_orders=[]
    \_isFitted=true
    \_isOrdered=true
    \_query=%[3]s:SliceQuery[0x70A1,0x70A2)
    \_vertices=1
    optimization                                                                               3.236
  WhereTraversalStep([WhereStartStep, EdgeVerte...                     1           1           0.160
    WhereStartStep                                                     1           1           0.028
    EdgeVertexStep(OUT)                                                1           1           0.034
    WhereEndStep(source)                                                                       0.019
  AddEdgeStep({~from=[[SelectOneStep(last,sourc...                                             0.000
    SelectOneStep(last,source)                                                                 0.000

Query may be rewritten like this

g.V().limit(1).as('source').coalesce(__.outE('%[3]s').where(otherV().is(g.V().limit(1).next())),addE('%[3]s').property("timestamp", System.currentTimeMillis()).to(g.V().limit(1).next())).profile()

And now query utilize VCI, edges between two vertices will be retrieved from single query without in memory filtering

JanusGraphStep(vertex,[])@[source]                                     1           1           0.870    39.83
  constructGraphCentricQuery                                                                   0.015
  constructGraphCentricQuery                                                                   0.008
  GraphCentricQuery                                                                            2.024
    \_condition=()
    \_orders=[]
    \_limit=1
    \_isFitted=false
    \_isOrdered=true
    \_query=[]
    scan                                                                                       0.053
    \_query=[]
    \_fullscan=true
    \_condition=VERTEX
CoalesceStep([[JanusGraphVertexStep([~adjacent....                     1           1           1.314    60.17
  JanusGraphVertexStep([~adjacent.eq(v[4160])])                        1           1           1.225
    \_condition=type[%[3]s]
    \_orders=[]
    \_isFitted=true
    \_isOrdered=true
    \_query=%[3]s:SliceQuery[0x70A0802040,0x70A0802041)
    \_vertices=1
    optimization                                                                               0.588
  AddEdgeStep({~to=[[ConstantStep(v[4160])]], l...                                             0.000
    ConstantStep(v[4160])                                                                      0.000

[rogaha]

I've uploaded the CPU flame from before and after the CPU spike.

[rogaha]

If I cannot find a solution for this problem, unfortunately, I'll have to replace Janusgraph with another Graph DB. I'm happy share more details or talk over Zoom/Meet/Teams if needed.