Generating index key for CacheInvalidationService markKeyAsExpiredInIndexStore

[jekramer]

I'm having trouble deciphering how to generate the key needed to invalidate an entry in the index store if I don't have a specific InternalRelation that I can call getIndexUpdates() for.

If I only know the vertex id that I want to be cleared from the index store, is it currently possible to generate relevant index keys? If so, how can I do that?

[porunov]

If I only know the vertex id that I want to be cleared from the index store, is it currently possible to generate relevant index keys? If so, how can I do that?

No, it's not possible to generate index keys knowing only vertex or edge id. Instead you need to know properties and their values which you want to expire in the index store.
I guess it could be possible to scan all index store values and invalidate related keys, but that would require quite a long scanning (depending on the index store size). Thus, I wasn't taking effort to implement that.
Instead you could call clearIndexStoreCache() as for now to fully invalidate all index store keys.

You can check a test method to see an example on how to generate relevant index keys using property name / values in the test class:

janusgraph/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java

Lines 8555 to 8583 in df62441

	private void invalidateUpdatedVertexProperty(StandardJanusGraph graph, long vertexIdUpdated, String propertyNameUpdated, Object previousPropertyValue, Object newPropertyValue, boolean withIndexConstraintsFilter){
	JanusGraphTransaction tx = graph.newTransaction();
	JanusGraphManagement graphMgmt = graph.openManagement();
	PropertyKey propertyKey = graphMgmt.getPropertyKey(propertyNameUpdated);
	CacheVertex cacheVertex = new CacheVertex((StandardJanusGraphTx) tx, vertexIdUpdated, ElementLifeCycle.Loaded);
	StandardVertexProperty propertyPreviousVal = new StandardVertexProperty(propertyKey.longId(), propertyKey, cacheVertex, previousPropertyValue, ElementLifeCycle.Removed);
	StandardVertexProperty propertyNewVal = new StandardVertexProperty(propertyKey.longId(), propertyKey, cacheVertex, newPropertyValue, ElementLifeCycle.New);
	IndexSerializer indexSerializer = graph.getIndexSerializer();
	Collection<IndexUpdate> indexUpdates;
	if(withIndexConstraintsFilter){
	indexUpdates = indexSerializer.getIndexUpdates(cacheVertex, Arrays.asList(propertyPreviousVal, propertyNewVal));
	} else {
	indexUpdates = indexSerializer.getIndexUpdatesNoConstraints(cacheVertex, Arrays.asList(propertyPreviousVal, propertyNewVal));
	}
	CacheInvalidationService invalidationService = graph.getDBCacheInvalidationService();
	for(IndexUpdate indexUpdate : indexUpdates){
	StaticBuffer keyToInvalidate = (StaticBuffer) indexUpdate.getKey();
	invalidationService.markKeyAsExpiredInIndexStore(keyToInvalidate);
	}
	invalidationService.forceClearExpiredKeysInIndexStoreCache();
	invalidationService.forceInvalidateVertexInEdgeStoreCache(vertexIdUpdated);
	graphMgmt.rollback();
	tx.rollback();
	}

In case you are interested to invalidate vertices / edges on all JanusGraph instances on their update or removal then you could implement it using a TransactionLog (https://docs.janusgraph.org/advanced-topics/transaction-log/). It will collect all relevant elements (with their properties) during add / update / remove operations and send them to that log. All interested JanusGraph instances could subscribe to that Change events, convert those Change events to relevant edge key (i.e. vertexId) and index keys (i.e. property name / values) and invalidate relevant edge and index store entries on each JanusGraph node.
The one thing you need to be aware of with TransactionLog is storage. It's not going to be cleared unless you explicitly add TTL and your storage backend supports it.
See configuration reference:

log.[X].ttl	Sets a TTL on all log entries, meaning that all entries added to this log expire after the configured amount of time. Requires that the log implementation supports TTL.

I guess other log implementations could be done using better fitted storages for that (like Kafka or similar).

At least I had this implementation idea for #3155 but didn't have a chance to work on that.

[porunov]

Thanks for the reply! Yep we were thinking of using Kafka for invalidating across instances. Is there anything existing to easily translate TransactionLog updates to a Kafka message, or is that something we would have to implement on our own? Just looking at docs a little, it seems like we could add a log processor that reads the change records, and we could implement that to forward on to Kafka?

JanusGraph provides you with the ability to use any implementation for logging but it doesn't come with pre-defined Kafka implementation or any other implementation except the default (org.janusgraph.diskstorage.log.kcvs.KCVSLogManager) storage implementation (i.e. Cassandra / HBase / etc.).

To add Kafka implementation you would need to provide a full class name via log.[X].backend configuration to the class which implements org.janusgraph.diskstorage.BackendException.LogManager interface. That class will have to have a public constructor which accept a single argument org.janusgraph.diskstorage.configuration.Configuration.

JanusGraph will instantiate that LogManager and provide it with the all configuration options provided via Configuration.

The main purpose of that LogManager is to open logs, where Log comes in form of the class implementing the org.janusgraph.diskstorage.log.Log interface. It can be any implementation, but I assume in your case it will be Kafka (which definitely makes sense).

Thus, in short you will need to:

• implement LogManager interface (see: KCVSLogManager for example)
• implement Log interface (see: KCVSLog for example)

KCVSLogManager by itself can be used instead of implementing LogManager from scratch, but for that you would need to implement the whole new Kafka backend storage for JanusGraph which could be more work then simply implementing LogManager only.

Unfortunately, we have poor documentation regarding log implementation and configuration, but you could use the logic above.
If you implement Kafka LogManager it would be great if you can contribute it to JanusGraph. I think many people would be happy to use Kafka as a log manager instead of their storage backend.

I'm a bit confused about how transaction details from one instance are useful for generating index keys on another instance. Do they just provide potential property/value options to check that may have index keys on the other instances?

In sort - yes.
When specific JanusGraph instance removes an element from the storage backend, that JanusGraph instance holds the latest information about that element (i.e. properties and relations). Using those properties and relations JanusGraph instance automatically invalidates all index entries which might potentially hold an item.
For example, let's think you have an index on name and you might have several index entries indexing different people with the same name. Removing / updating name for a single person with name Bob requires to invalidate all index entries which may answer this index.
Assume you have 3 JanusGraph instances. Each instance cached a result for g.V().has("name", "Bob"). Let's assume a result consists of 5 vertices (i.e. there are 5 people with name = Bob in your database).
Now you removed one vertex (i.e. one person) with name = Bob but 4 other left. It means that g.V().has("name", "Bob") returning 5 elements isn't valid anymore because it contains stale data. Now we need to invalidate all index entries answering g.V().has("name", "Bob") on all instances. As I said above, an instance which processed a deletion will hold all the information about the deleted elements. Thus, it automatically invalidates g.V().has("name", "Bob"). That said, 2 other JanusGraph instances are not aware of this removal and cannot invalidate the affected index entries without knowing what exactly was removed.
That's exactly what TransactionLog provides us. It can be used to pass that information to all other interested JanusGraph instances (i.e. all instances in this case) and invalidate potential stale index entries.
Of course, it doesn't guarantee that those JanusGraph instances had Bob cached anywhere, but there is no way of knowing, thus we should still trigger such invalidation.

[porunov]

@jekramer I have created PR for ChangeProcessor implementation (which allows invalidating cache correctly). It's not finished yet, but you may find an example of ChangeProcessor to use there: #3581
That change processor has a downside because it doesn't currently work correctly when an updated element was removed before it's change log was processed. To fix it we should probably wrap it into some cached or wrapper element to not execute an actual query on a removed element. Moreover, we don't need to execute anything because ChangeProcessor already has all information it needs for invalidation.

[jekramer]

thank you for this, it will be a huge help! I had just started to look into ChangeProcessors the other day. I'm also wondering if there's any sort of sender ID that would allow us to skip processing on transactions from the same instance, since we know those entries will have already been invalidated locally?

[porunov]

thank you for this, it will be a huge help! I had just started to look into ChangeProcessors the other day. I'm also wondering if there's any sort of sender ID that would allow us to skip processing on transactions from the same instance, since we know those entries will have already been invalidated locally?

Good optimization idea!
Yes, it's passed into the processor as well. See documentation about it here.
From the documentation:
The provided transaction id can be used to investigate the origin of the transaction which is uniquely identified by the combination of the id of the JanusGraph instance that executed the transaction (txId.getInstanceId()) and the instance specific transaction id (txId.getTransactionId()). In addition, the time of the transaction is available through txId.getTransactionTime().

[jekramer]

Ahh perfect, thanks!

[jekramer]

hey @porunov, this is working well so far! I just want to clarify which log sub-namespace I should be setting ttl for. Based on the transaction log documentation, it seems like it should be user, but I also noticed in the JanusGraph code that there is a tx namespace too. Is tx the more internal "transaction" log described here, and what I'm interacting with is the user "trigger" log?

[jekramer]

figured this out with testing after I posted here. log.user.read-lag-time works with my processor, so I'm using log.user.ttl for expiration