aboutsummaryrefslogtreecommitdiffstats
path: root/champ-lib/champ-janus/src/main
diff options
context:
space:
mode:
authormichaere <michaere@amdocs.com>2018-03-05 16:33:32 +0000
committermichaere <michaere@amdocs.com>2018-03-07 11:17:22 +0000
commitc74f7b13b573386e70c10721fc391624ee792ed6 (patch)
treeb44995474ff938b4b03c9b234f95b71bc75d6b79 /champ-lib/champ-janus/src/main
parent9fc28cff11a4b570618c0f533ce9de6209a5dd0c (diff)
Port champ-microservice project restructure
Includes project restructure and introduction of a parent pom. The original source folder and core functionality is now held within champ-lib, with champ-service forming the ajsc microservice from which it injects champ-lib core functionality. Issue-ID: AAI-813 Change-Id: I2ce0c4a70e485665276e7955572de23969deb706 Signed-off-by: michaere <michaere@amdocs.com>
Diffstat (limited to 'champ-lib/champ-janus/src/main')
-rw-r--r--champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/GraphSON.java66
-rw-r--r--champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/JanusChampGraphImpl.java493
2 files changed, 559 insertions, 0 deletions
diff --git a/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/GraphSON.java b/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/GraphSON.java
new file mode 100644
index 0000000..9e849b8
--- /dev/null
+++ b/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/GraphSON.java
@@ -0,0 +1,66 @@
+/**
+ * ============LICENSE_START==========================================
+ * org.onap.aai
+ * ===================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017 Amdocs
+ * ===================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END============================================
+ * ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ */
+package org.onap.aai.champjanus.graph.impl;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+// import com.thinkaurelius.titan.graphdb.tinkerpop.TitanIoRegistry;
+import org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONMapper;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONWriter;
+import org.onap.aai.champcore.FormatMapper;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class GraphSON implements FormatMapper {
+ private final GraphSONMapper mapper;
+ private final GraphSONWriter writer;
+ protected JsonParser parser;
+
+ public GraphSON() {
+ this.mapper = GraphSONMapper.build().addRegistry(JanusGraphIoRegistry.getInstance ()).create();
+ this.writer = GraphSONWriter.build().mapper(this.mapper).create();
+ this.parser = new JsonParser();
+ }
+
+ public JsonObject formatObject(Object v) {
+ OutputStream os = new ByteArrayOutputStream();
+ String result = "";
+
+ try {
+ this.writer.writeVertex(os, (Vertex)v, Direction.BOTH);
+ result = os.toString();
+ } catch (IOException var5) {
+ var5.printStackTrace();
+ }
+
+ return this.parser.parse(result).getAsJsonObject();
+ }
+
+ public int parallelThreshold() {
+ return 50;
+ }
+}
diff --git a/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/JanusChampGraphImpl.java b/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/JanusChampGraphImpl.java
new file mode 100644
index 0000000..9ede7df
--- /dev/null
+++ b/champ-lib/champ-janus/src/main/java/org/onap/aai/champjanus/graph/impl/JanusChampGraphImpl.java
@@ -0,0 +1,493 @@
+/**
+ * ============LICENSE_START==========================================
+ * org.onap.aai
+ * ===================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017 Amdocs
+ * ===================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END============================================
+ * ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ */
+package org.onap.aai.champjanus.graph.impl;
+
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.janusgraph.core.*;
+import org.onap.aai.champcore.Formatter;
+import org.janusgraph.core.schema.JanusGraphIndex;
+import org.janusgraph.core.schema.JanusGraphManagement;
+import org.janusgraph.core.schema.SchemaAction;
+import org.janusgraph.core.schema.SchemaStatus;
+import org.janusgraph.graphdb.database.management.ManagementSystem;
+import org.onap.aai.champcore.ChampCapabilities;
+import org.onap.aai.champcore.FormatMapper;
+import org.onap.aai.champcore.exceptions.ChampIndexNotExistsException;
+import org.onap.aai.champcore.exceptions.ChampSchemaViolationException;
+import org.onap.aai.champcore.graph.impl.AbstractTinkerpopChampGraph;
+import org.onap.aai.champcore.model.*;
+import org.onap.aai.champcore.schema.ChampSchemaEnforcer;
+import org.onap.aai.champcore.schema.DefaultChampSchemaEnforcer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.temporal.ChronoUnit;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+public final class JanusChampGraphImpl extends AbstractTinkerpopChampGraph {
+ private static final Logger LOGGER = LoggerFactory.getLogger(JanusChampGraphImpl.class);
+ private static final String JANUS_CASSANDRA_KEYSPACE = "storage.cassandra.keyspace";
+ private static final String JANUS_HBASE_TABLE = "storage.hbase.table";
+ private static final String JANUS_UNIQUE_SUFFIX = "graph.unique-instance-id-suffix";
+ private static final ChampSchemaEnforcer SCHEMA_ENFORCER = new DefaultChampSchemaEnforcer();
+ private static final int REGISTER_OBJECT_INDEX_TIMEOUT_SECS = 30;
+
+ private static final ChampCapabilities CAPABILITIES = new ChampCapabilities() {
+
+ @Override
+ public boolean canDeleteObjectIndices() {
+ return false;
+ }
+
+ @Override
+ public boolean canDeleteRelationshipIndices() {
+ return false;
+ }
+ };
+
+ private final JanusGraph graph;
+
+ public JanusChampGraphImpl(Builder builder) {
+ super(builder.graphConfiguration);
+ final JanusGraphFactory.Builder janusGraphBuilder = JanusGraphFactory.build();
+
+ for (Map.Entry<String, Object> janusGraphProperty : builder.graphConfiguration.entrySet()) {
+ janusGraphBuilder.set(janusGraphProperty.getKey(), janusGraphProperty.getValue());
+ }
+
+ janusGraphBuilder.set(JANUS_UNIQUE_SUFFIX, ((short) new Random().nextInt(Short.MAX_VALUE)+""));
+
+ final Object storageBackend = builder.graphConfiguration.get("storage.backend");
+
+ if ("cassandra".equals(storageBackend) ||
+ "cassandrathrift".equals(storageBackend) ||
+ "astyanax".equals(storageBackend) ||
+ "embeddedcassandra".equals(storageBackend)) {
+
+ janusGraphBuilder.set(JANUS_CASSANDRA_KEYSPACE, builder.graphName);
+ } else if ("hbase".equals(storageBackend)) {
+ janusGraphBuilder.set(JANUS_HBASE_TABLE, builder.graphName);
+ } else if ("berkleyje".equals(storageBackend)) {
+ throw new RuntimeException("storage.backend=berkleyje cannot handle multiple graphs on a single DB, not usable");
+ } else if ("inmemory".equals(storageBackend)) {
+ } else {
+ throw new RuntimeException("Unknown storage.backend=" + storageBackend);
+ }
+
+ LOGGER.info("Instantiated data access layer for Janus graph data store with backend: " + storageBackend);
+ this.graph = janusGraphBuilder.open();
+ }
+
+ public static class Builder {
+ private final String graphName;
+
+ private final Map<String, Object> graphConfiguration = new HashMap<String, Object>();
+
+ public Builder(String graphName) {
+ this.graphName = graphName;
+ }
+
+ public Builder(String graphName, Map<String, Object> properties) {
+ this.graphName = graphName;
+ properties(properties);
+ }
+
+ public Builder properties(Map<String, Object> properties) {
+ if (properties.containsKey(JANUS_CASSANDRA_KEYSPACE)) {
+ throw new IllegalArgumentException("Cannot use path " + JANUS_CASSANDRA_KEYSPACE
+ + " in initial configuration - this path is used"
+ + " to specify graph names");
+ }
+
+ this.graphConfiguration.putAll(properties);
+ return this;
+ }
+
+ public Builder property(String path, Object value) {
+ if (path.equals(JANUS_CASSANDRA_KEYSPACE)) {
+ throw new IllegalArgumentException("Cannot use path " + JANUS_CASSANDRA_KEYSPACE
+ + " in initial configuration - this path is used"
+ + " to specify graph names");
+ }
+ graphConfiguration.put(path, value);
+ return this;
+ }
+
+ public JanusChampGraphImpl build() {
+ return new JanusChampGraphImpl(this);
+ }
+ }
+
+ @Override
+ protected JanusGraph getGraph() {
+ return graph;
+ }
+
+
+ @Override
+ protected ChampSchemaEnforcer getSchemaEnforcer() {
+ return SCHEMA_ENFORCER;
+ }
+
+ @Override
+ public void executeStoreObjectIndex(ChampObjectIndex index) {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call storeObjectIndex() after shutdown has been initiated");
+ }
+
+ final JanusGraph graph = getGraph();
+ final JanusGraphManagement createIndexMgmt = graph.openManagement();
+ final PropertyKey pk = createIndexMgmt.getOrCreatePropertyKey(index.getField().getName());
+
+ if (createIndexMgmt.getGraphIndex(index.getName()) != null) {
+ createIndexMgmt.rollback();
+ return; //Ignore, index already exists
+ }
+
+ createIndexMgmt.buildIndex(index.getName(), Vertex.class).addKey(pk).buildCompositeIndex();
+
+ createIndexMgmt.commit();
+ graph.tx().commit();
+
+ awaitIndexCreation(index.getName());
+ }
+
+ @Override
+ public Optional<ChampObjectIndex> retrieveObjectIndex(String indexName) {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call retrieveObjectIndex() after shutdown has been initiated");
+ }
+
+ final JanusGraphManagement retrieveIndexMgmt = getGraph().openManagement();
+ final JanusGraphIndex index = retrieveIndexMgmt.getGraphIndex(indexName);
+
+ if (index == null) {
+ return Optional.empty();
+ }
+ if (index.getIndexedElement() != JanusGraphVertex.class) {
+ return Optional.empty();
+ }
+
+ return Optional.of(ChampObjectIndex.create()
+ .ofName(indexName)
+ .onType(ChampObject.ReservedTypes.ANY.toString())
+ .forField(index.getFieldKeys()[0].name())
+ .build());
+ }
+
+ @Override
+ public Stream<ChampObjectIndex> retrieveObjectIndices() {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call retrieveObjectIndices() after shutdown has been initiated");
+ }
+
+ final JanusGraphManagement createIndexMgmt = getGraph().openManagement();
+ final Iterator<JanusGraphIndex> indices = createIndexMgmt.getGraphIndexes(Vertex.class).iterator();
+
+ final Iterator<ChampObjectIndex> objIter = new Iterator<ChampObjectIndex>() {
+
+ private ChampObjectIndex next;
+
+ @Override
+ public boolean hasNext() {
+ if (indices.hasNext()) {
+ final JanusGraphIndex index = indices.next();
+
+ next = ChampObjectIndex.create()
+ .ofName(index.name())
+ .onType(ChampObject.ReservedTypes.ANY.toString())
+ .forField(index.getFieldKeys()[0].name())
+ .build();
+ return true;
+ }
+
+ next = null;
+ return false;
+ }
+
+ @Override
+ public ChampObjectIndex next() {
+ if (next == null) {
+ throw new NoSuchElementException();
+ }
+
+ return next;
+ }
+ };
+
+ return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
+ objIter, Spliterator.ORDERED | Spliterator.NONNULL), false);
+ }
+
+ @Override
+ public void executeDeleteObjectIndex(String indexName) throws ChampIndexNotExistsException {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call deleteObjectIndex() after shutdown has been initiated");
+ }
+
+ throw new UnsupportedOperationException("Cannot delete indices using the JanusChampImpl");
+ }
+
+ @Override
+ public void executeStoreRelationshipIndex(ChampRelationshipIndex index) {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call storeRelationshipIndex() after shutdown has been initiated");
+ }
+
+ final JanusGraph graph = getGraph();
+ final JanusGraphManagement createIndexMgmt = graph.openManagement();
+ final PropertyKey pk = createIndexMgmt.getOrCreatePropertyKey(index.getField().getName());
+
+ if (createIndexMgmt.getGraphIndex(index.getName()) != null) {
+ return; //Ignore, index already exists
+ }
+ createIndexMgmt.buildIndex(index.getName(), Edge.class).addKey(pk).buildCompositeIndex();
+
+ createIndexMgmt.commit();
+ graph.tx().commit();
+
+ awaitIndexCreation(index.getName());
+ }
+
+ @Override
+ public Optional<ChampRelationshipIndex> retrieveRelationshipIndex(String indexName) {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call retrieveRelationshipIndex() after shutdown has been initiated");
+ }
+
+ final JanusGraphManagement retrieveIndexMgmt = getGraph().openManagement();
+ final JanusGraphIndex index = retrieveIndexMgmt.getGraphIndex(indexName);
+
+ if (index == null) {
+ return Optional.empty();
+ }
+ if (index.getIndexedElement() != JanusGraphEdge.class) {
+ return Optional.empty();
+ }
+
+ return Optional.of(ChampRelationshipIndex.create()
+ .ofName(indexName)
+ .onType(ChampObject.ReservedTypes.ANY.toString())
+ .forField(index.getFieldKeys()[0].name())
+ .build());
+ }
+
+ @Override
+ public Stream<ChampRelationshipIndex> retrieveRelationshipIndices() {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call retrieveRelationshipIndices() after shutdown has been initiated");
+ }
+
+ final JanusGraphManagement createIndexMgmt = getGraph().openManagement();
+ final Iterator<JanusGraphIndex> indices = createIndexMgmt.getGraphIndexes(Edge.class).iterator();
+
+ final Iterator<ChampRelationshipIndex> objIter = new Iterator<ChampRelationshipIndex>() {
+
+ private ChampRelationshipIndex next;
+
+ @Override
+ public boolean hasNext() {
+ if (indices.hasNext()) {
+ final JanusGraphIndex index = indices.next();
+
+ next = ChampRelationshipIndex.create()
+ .ofName(index.name())
+ .onType(ChampRelationship.ReservedTypes.ANY.toString())
+ .forField(index.getFieldKeys()[0].name())
+ .build();
+ return true;
+ }
+
+ next = null;
+ return false;
+ }
+
+ @Override
+ public ChampRelationshipIndex next() {
+ if (next == null) {
+ throw new NoSuchElementException();
+ }
+
+ return next;
+ }
+ };
+
+ return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
+ objIter, Spliterator.ORDERED | Spliterator.NONNULL), false);
+ }
+
+ @Override
+ public void executeDeleteRelationshipIndex(String indexName) throws ChampIndexNotExistsException {
+ if (isShutdown()) {
+ throw new IllegalStateException("Cannot call deleteRelationshipIndex() after shutdown has been initiated");
+ }
+
+ throw new UnsupportedOperationException("Cannot delete indices using the JanusChampImpl");
+ }
+
+ private Cardinality getJanusCardinality(ChampCardinality cardinality) {
+ switch (cardinality) {
+ case LIST:
+ return Cardinality.LIST;
+ case SET:
+ return Cardinality.SET;
+ case SINGLE:
+ return Cardinality.SINGLE;
+ default:
+ throw new RuntimeException("Unknown ChampCardinality " + cardinality);
+ }
+ }
+
+ private void awaitIndexCreation(String indexName) {
+ //Wait for the index to become available
+ try {
+ if (ManagementSystem.awaitGraphIndexStatus(graph, indexName)
+ .status(SchemaStatus.ENABLED)
+ .timeout(1, ChronoUnit.SECONDS)
+ .call()
+ .getSucceeded()) {
+ return; //Empty graphs immediately ENABLE indices
+ }
+
+ if (!ManagementSystem.awaitGraphIndexStatus(graph, indexName)
+ .status(SchemaStatus.REGISTERED)
+ .timeout(REGISTER_OBJECT_INDEX_TIMEOUT_SECS, ChronoUnit.SECONDS)
+ .call()
+ .getSucceeded()) {
+ LOGGER.warn("Object index was created, but timed out while waiting for it to be registered");
+ return;
+ }
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while waiting for object index creation status");
+ return;
+ }
+
+ //Reindex the existing data
+
+ try {
+ final JanusGraphManagement updateIndexMgmt = graph.openManagement();
+ updateIndexMgmt.updateIndex(updateIndexMgmt.getGraphIndex(indexName), SchemaAction.REINDEX).get();
+ updateIndexMgmt.commit();
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while reindexing for object index");
+ return;
+ } catch (ExecutionException e) {
+ LOGGER.warn("Exception occurred during reindexing procedure for creating object index " + indexName, e);
+ }
+
+ try {
+ ManagementSystem.awaitGraphIndexStatus(graph, indexName)
+ .status(SchemaStatus.ENABLED)
+ .timeout(10, ChronoUnit.MINUTES)
+ .call();
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while waiting for index to transition to ENABLED state");
+ return;
+ }
+ }
+
+ @Override
+ public ChampCapabilities capabilities() {
+ return CAPABILITIES;
+ }
+
+ public void storeSchema(ChampSchema schema) throws ChampSchemaViolationException {
+ if (isShutdown()) throw new IllegalStateException("Cannot call storeSchema() after shutdown has been initiated");
+
+ final ChampSchema currentSchema = retrieveSchema();
+ final JanusGraphManagement mgmt = getGraph().openManagement();
+
+ try {
+ for (ChampObjectConstraint objConstraint : schema.getObjectConstraints().values()) {
+ for (ChampPropertyConstraint propConstraint : objConstraint.getPropertyConstraints()) {
+ final Optional<ChampObjectConstraint> currentObjConstraint = currentSchema.getObjectConstraint(objConstraint.getType());
+
+ if (currentObjConstraint.isPresent()) {
+ final Optional<ChampPropertyConstraint> currentPropConstraint = currentObjConstraint.get().getPropertyConstraint(propConstraint.getField().getName());
+
+ if (currentPropConstraint.isPresent() && currentPropConstraint.get().compareTo(propConstraint) != 0) {
+ throw new ChampSchemaViolationException("Cannot update already existing property on object type " + objConstraint.getType() + ": " + propConstraint);
+ }
+ }
+
+ final String newPropertyKeyName = propConstraint.getField().getName();
+
+ if (mgmt.getPropertyKey(newPropertyKeyName) != null) continue; //Check Janus to see if another node created this property key
+
+ mgmt.makePropertyKey(newPropertyKeyName)
+ .dataType(propConstraint.getField().getJavaType())
+ .cardinality(getJanusCardinality(propConstraint.getCardinality()))
+ .make();
+ }
+ }
+
+ for (ChampRelationshipConstraint relConstraint : schema.getRelationshipConstraints().values()) {
+
+ final Optional<ChampRelationshipConstraint> currentRelConstraint = currentSchema.getRelationshipConstraint(relConstraint.getType());
+
+ for (ChampPropertyConstraint propConstraint : relConstraint.getPropertyConstraints()) {
+
+ if (currentRelConstraint.isPresent()) {
+ final Optional<ChampPropertyConstraint> currentPropConstraint = currentRelConstraint.get().getPropertyConstraint(propConstraint.getField().getName());
+
+ if (currentPropConstraint.isPresent() && currentPropConstraint.get().compareTo(propConstraint) != 0) {
+ throw new ChampSchemaViolationException("Cannot update already existing property on relationship type " + relConstraint.getType());
+ }
+ }
+
+ final String newPropertyKeyName = propConstraint.getField().getName();
+
+ if (mgmt.getPropertyKey(newPropertyKeyName) != null) continue; //Check Janus to see if another node created this property key
+
+ mgmt.makePropertyKey(newPropertyKeyName)
+ .dataType(propConstraint.getField().getJavaType())
+ .cardinality(getJanusCardinality(propConstraint.getCardinality()))
+ .make();
+ }
+
+ final EdgeLabel edgeLabel = mgmt.getEdgeLabel(relConstraint.getType());
+
+ if (edgeLabel != null) {
+ mgmt.makeEdgeLabel(relConstraint.getType())
+ .directed()
+ .make();
+ }
+ }
+
+ mgmt.commit();
+
+ super.storeSchema(schema);
+ } catch (SchemaViolationException | ChampSchemaViolationException e) {
+ mgmt.rollback();
+ throw new ChampSchemaViolationException(e);
+ }
+ }
+
+ public GraphTraversal<?, ?> hasLabel(GraphTraversal<?, ?> query, Object type) {
+ return query.hasLabel(type);
+ }
+} \ No newline at end of file