Skip to content
This repository has been archived by the owner on Dec 13, 2023. It is now read-only.

Commit

Permalink
Merge pull request #1526 from Netflix/cass_metadata_dao
Browse files Browse the repository at this point in the history 
implemented metadata and eventHandler dao in cassandra module
  • Loading branch information
@apanicker-nflx
apanicker-nflx committed Feb 22, 2020
2 parents 1f6bad2 + 6a635e2 commit b513405
Show file tree
Hide file tree
Showing 13 changed files with 1,285 additions and 168 deletions.
6 changes: 4 additions & 2 deletions cassandra-persistence/README.md
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
### Note
This provides a partial implementation of the ExecutionDAO using Cassandra as the datastore.
This module provides a partial implementation of the ExecutionDAO using Cassandra as the datastore.
The execution data is stored in Cassandra in the `workflows` table. A task to workflow mapping is also maintained in a separate `task_lookup` table.

All datastore operations that are used during the critical execution path of a workflow are currently implemented. This includes CRUD operations for workflows and tasks.

This does not provide implementations for the QueueDAO and MetadataDAO interfaces.
This module provides complete implementations for MetadataDAO and EventHandlerDAO interfaces.

This module does not provide implementations for the QueueDAO and PollDataDAO interfaces.

10 changes: 8 additions & 2 deletions cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/CassandraModule.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2016 Netflix, Inc.
* Copyright 2020 Netflix, Inc.
*
* 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
Expand All @@ -15,8 +15,12 @@
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import com.google.inject.AbstractModule;
import com.netflix.conductor.dao.cassandra.CassandraExecutionDAO;
import com.netflix.conductor.dao.EventHandlerDAO;
import com.netflix.conductor.dao.ExecutionDAO;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.dao.cassandra.CassandraEventHandlerDAO;
import com.netflix.conductor.dao.cassandra.CassandraExecutionDAO;
import com.netflix.conductor.dao.cassandra.CassandraMetadataDAO;

public class CassandraModule extends AbstractModule {

Expand All @@ -26,6 +30,8 @@ protected void configure() {
bind(Cluster.class).toProvider(CassandraClusterProvider.class).asEagerSingleton();
bind(Session.class).toProvider(CassandraSessionProvider.class);

bind(MetadataDAO.class).to(CassandraMetadataDAO.class);
bind(ExecutionDAO.class).to(CassandraExecutionDAO.class);
bind(EventHandlerDAO.class).to(CassandraEventHandlerDAO.class);
}
}
156 changes: 107 additions & 49 deletions ...andra-persistence/src/main/java/com/netflix/conductor/dao/cassandra/CassandraBaseDAO.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,31 @@

import static com.netflix.conductor.util.Constants.DAO_NAME;
import static com.netflix.conductor.util.Constants.ENTITY_KEY;
import static com.netflix.conductor.util.Constants.EVENT_HANDLER_KEY;
import static com.netflix.conductor.util.Constants.EVENT_HANDLER_NAME_KEY;
import static com.netflix.conductor.util.Constants.HANDLERS_KEY;
import static com.netflix.conductor.util.Constants.PAYLOAD_KEY;
import static com.netflix.conductor.util.Constants.SHARD_ID_KEY;
import static com.netflix.conductor.util.Constants.TABLE_EVENT_HANDLERS;
import static com.netflix.conductor.util.Constants.TABLE_TASK_DEFS;
import static com.netflix.conductor.util.Constants.TABLE_TASK_DEF_LIMIT;
import static com.netflix.conductor.util.Constants.TABLE_TASK_LOOKUP;
import static com.netflix.conductor.util.Constants.TABLE_WORKFLOWS;
import static com.netflix.conductor.util.Constants.TABLE_WORKFLOW_DEFS;
import static com.netflix.conductor.util.Constants.TABLE_WORKFLOW_DEFS_INDEX;
import static com.netflix.conductor.util.Constants.TASK_DEFINITION_KEY;
import static com.netflix.conductor.util.Constants.TASK_DEFS_KEY;
import static com.netflix.conductor.util.Constants.TASK_DEF_NAME_KEY;
import static com.netflix.conductor.util.Constants.TASK_ID_KEY;
import static com.netflix.conductor.util.Constants.TOTAL_PARTITIONS_KEY;
import static com.netflix.conductor.util.Constants.TOTAL_TASKS_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_DEFINITION_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_DEF_INDEX_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_DEF_INDEX_VALUE;
import static com.netflix.conductor.util.Constants.WORKFLOW_DEF_NAME_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_DEF_NAME_VERSION_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_ID_KEY;
import static com.netflix.conductor.util.Constants.WORKFLOW_VERSION_KEY;

import com.datastax.driver.core.DataType;
import com.datastax.driver.core.Session;
Expand All @@ -44,41 +59,38 @@
/**
* Creates the keyspace and tables.
* <p>
* CREATE KEYSPACE IF NOT EXISTS conductor
* WITH replication = { 'class' : 'NetworkTopologyStrategy', 'us-east': '3'};
* CREATE KEYSPACE IF NOT EXISTS conductor WITH replication = { 'class' : 'NetworkTopologyStrategy', 'us-east': '3'};
* <p>
* CREATE TABLE IF NOT EXISTS conductor.workflows (
* workflow_id uuid,
* shard_id int,
* task_id text,
* entity text,
* payload text,
* total_tasks int STATIC,
* total_partitions int STATIC,
* PRIMARY KEY((workflow_id, shard_id), entity, task_id)
* );
* CREATE TABLE IF NOT EXISTS conductor.workflows ( workflow_id uuid, shard_id int, task_id text, entity text, payload
* text, total_tasks int STATIC, total_partitions int STATIC, PRIMARY KEY((workflow_id, shard_id), entity, task_id) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.task_lookup(
* task_id uuid,
* workflow_id uuid,
* PRIMARY KEY (task_id)
* );
* CREATE TABLE IF NOT EXISTS conductor.task_lookup( task_id uuid, workflow_id uuid, PRIMARY KEY (task_id) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.task_def_limit(
* task_def_name text,
* task_id uuid,
* workflow_id uuid,
* PRIMARY KEY ((task_def_name), task_id_key)
* );
* CREATE TABLE IF NOT EXISTS conductor.task_def_limit( task_def_name text, task_id uuid, workflow_id uuid, PRIMARY KEY
* ((task_def_name), task_id_key) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.workflow_definitions( workflow_def_name text, version int, workflow_definition
* text, PRIMARY KEY ((workflow_def_name), version) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.workflow_defs_index( workflow_def_version_index text, workflow_def_name_version
* text, workflow_def_index_value text,PRIMARY KEY ((workflow_def_version_index), workflow_def_name_version) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.task_definitions( task_defs text, task_def_name text, task_definition text,
* PRIMARY KEY ((task_defs), task_def_name) );
* <p>
* CREATE TABLE IF NOT EXISTS conductor.event_handlers( handlers text, event_handler_name text, event_handler text,
* PRIMARY KEY ((handlers), event_handler_name) );
*/
public class CassandraBaseDAO {

private static final Logger LOGGER = LoggerFactory.getLogger(CassandraBaseDAO.class);

private final ObjectMapper objectMapper;

protected final Session session;
protected final CassandraConfiguration config;

private boolean initialized = false;

public CassandraBaseDAO(Session session, ObjectMapper objectMapper, CassandraConfiguration config) {
this.session = session;
this.objectMapper = objectMapper;
Expand All @@ -89,11 +101,18 @@ public CassandraBaseDAO(Session session, ObjectMapper objectMapper, CassandraCon

private void init() {
try {
session.execute(getCreateKeyspaceStatement());
session.execute(getCreateWorkflowsTableStatement());
session.execute(getCreateTaskLookupTableStatement());
session.execute(getCreateTaskDefLimitStatement());
LOGGER.info("CassandraDAO initialization complete! Tables created!");
if (!initialized) {
session.execute(getCreateKeyspaceStatement());
session.execute(getCreateWorkflowsTableStatement());
session.execute(getCreateTaskLookupTableStatement());
session.execute(getCreateTaskDefLimitTableStatement());
session.execute(getCreateWorkflowDefsTableStatement());
session.execute(getCreateWorkflowDefsIndexTableStatement());
session.execute(getCreateTaskDefsTableStatement());
session.execute(getCreateEventHandlersTableStatement());
LOGGER.info("CassandraDAO initialization complete! Tables created!");
initialized = true;
}
} catch (Exception e) {
LOGGER.error("Error initializing and setting up keyspace and table in cassandra", e);
throw e;
Expand All @@ -102,35 +121,36 @@ private void init() {

private String getCreateKeyspaceStatement() {
return SchemaBuilder.createKeyspace(config.getCassandraKeyspace())
.ifNotExists()
.with()
.replication(ImmutableMap.of("class", config.getReplicationStrategy(), config.getReplicationFactorKey(), config.getReplicationFactorValue()))
.durableWrites(true)
.getQueryString();
.ifNotExists()
.with()
.replication(ImmutableMap.of("class", config.getReplicationStrategy(), config.getReplicationFactorKey(),
config.getReplicationFactorValue()))
.durableWrites(true)
.getQueryString();
}

private String getCreateWorkflowsTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_WORKFLOWS)
.ifNotExists()
.addPartitionKey(WORKFLOW_ID_KEY, DataType.uuid())
.addPartitionKey(SHARD_ID_KEY, DataType.cint())
.addClusteringColumn(ENTITY_KEY, DataType.text())
.addClusteringColumn(TASK_ID_KEY, DataType.text())
.addColumn(PAYLOAD_KEY, DataType.text())
.addStaticColumn(TOTAL_TASKS_KEY, DataType.cint())
.addStaticColumn(TOTAL_PARTITIONS_KEY, DataType.cint())
.getQueryString();
.ifNotExists()
.addPartitionKey(WORKFLOW_ID_KEY, DataType.uuid())
.addPartitionKey(SHARD_ID_KEY, DataType.cint())
.addClusteringColumn(ENTITY_KEY, DataType.text())
.addClusteringColumn(TASK_ID_KEY, DataType.text())
.addColumn(PAYLOAD_KEY, DataType.text())
.addStaticColumn(TOTAL_TASKS_KEY, DataType.cint())
.addStaticColumn(TOTAL_PARTITIONS_KEY, DataType.cint())
.getQueryString();
}

private String getCreateTaskLookupTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_TASK_LOOKUP)
.ifNotExists()
.addPartitionKey(TASK_ID_KEY, DataType.uuid())
.addColumn(WORKFLOW_ID_KEY, DataType.uuid())
.getQueryString();
.ifNotExists()
.addPartitionKey(TASK_ID_KEY, DataType.uuid())
.addColumn(WORKFLOW_ID_KEY, DataType.uuid())
.getQueryString();
}

private String getCreateTaskDefLimitStatement() {
private String getCreateTaskDefLimitTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_TASK_DEF_LIMIT)
.ifNotExists()
.addPartitionKey(TASK_DEF_NAME_KEY, DataType.text())
Expand All @@ -139,6 +159,42 @@ private String getCreateTaskDefLimitStatement() {
.getQueryString();
}

private String getCreateWorkflowDefsTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_WORKFLOW_DEFS)
.ifNotExists()
.addPartitionKey(WORKFLOW_DEF_NAME_KEY, DataType.text())
.addClusteringColumn(WORKFLOW_VERSION_KEY, DataType.cint())
.addColumn(WORKFLOW_DEFINITION_KEY, DataType.text())
.getQueryString();
}

private String getCreateWorkflowDefsIndexTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_WORKFLOW_DEFS_INDEX)
.ifNotExists()
.addPartitionKey(WORKFLOW_DEF_INDEX_KEY, DataType.text())
.addClusteringColumn(WORKFLOW_DEF_NAME_VERSION_KEY, DataType.text())
.addColumn(WORKFLOW_DEF_INDEX_VALUE, DataType.text())
.getQueryString();
}

private String getCreateTaskDefsTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_TASK_DEFS)
.ifNotExists()
.addPartitionKey(TASK_DEFS_KEY, DataType.text())
.addClusteringColumn(TASK_DEF_NAME_KEY, DataType.text())
.addColumn(TASK_DEFINITION_KEY, DataType.text())
.getQueryString();
}

private String getCreateEventHandlersTableStatement() {
return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_EVENT_HANDLERS)
.ifNotExists()
.addPartitionKey(HANDLERS_KEY, DataType.text())
.addClusteringColumn(EVENT_HANDLER_NAME_KEY, DataType.text())
.addColumn(EVENT_HANDLER_KEY, DataType.text())
.getQueryString();
}

String toJson(Object value) {
try {
return objectMapper.writeValueAsString(value);
Expand Down Expand Up @@ -168,10 +224,12 @@ void recordCassandraDaoEventRequests(String action, String event) {
}

void recordCassandraDaoPayloadSize(String action, int size, String taskType, String workflowType) {
Monitors.recordDaoPayloadSize(DAO_NAME, action, StringUtils.defaultIfBlank(taskType, ""), StringUtils.defaultIfBlank(workflowType, ""), size);
Monitors.recordDaoPayloadSize(DAO_NAME, action, StringUtils.defaultIfBlank(taskType, ""),
StringUtils.defaultIfBlank(workflowType, ""), size);
}

static class WorkflowMetadata {

private int totalTasks;
private int totalPartitions;

Expand Down

0 comments on commit b513405

Please sign in to comment.