diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -121,7 +121,7 @@ ``` storage: - cls: local + cls: postgresql db: "dbname=softwareheritage-dev user= password=" objstorage: cls: pathslicing @@ -187,14 +187,37 @@ url: http://localhost:5002/ ``` -You could directly define a local storage with the following snippet: +You could directly define a postgresql storage with the following snippet: ``` storage: - cls: local + cls: postgresql db: service=swh-dev objstorage: cls: pathslicing root: /home/storage/swh-storage/ slicing: 0:2/2:4/4:6 ``` + +## Cassandra + +As an alternative to PostgreSQL, swh-storage can use Cassandra as a database backend. +It can be used like this: + +``` +storage: + cls: cassandra + hosts: + - localhost + objstorage: + cls: pathslicing + root: /home/storage/swh-storage/ + slicing: 0:2/2:4/4:6 +``` + +The Cassandra swh-storage implementation supports both Cassandra >= 4.0-alpha2 +and ScyllaDB >= 4.4 (and possibly earlier versions, but this is untested). + +While the main code supports both transparently, running tests +or configuring the schema requires specific code when using ScyllaDB, +enabled with the `SWH_USE_SCYLLADB=1` configuration variable. diff --git a/swh/storage/cassandra/cql.py b/swh/storage/cassandra/cql.py --- a/swh/storage/cassandra/cql.py +++ b/swh/storage/cassandra/cql.py @@ -980,7 +980,9 @@ return next(self.origin_visit_status_get(origin, visit), None) @_prepared_select_statement( - OriginVisitStatusRow, "WHERE origin = ? AND visit = ? ORDER BY date DESC" + OriginVisitStatusRow, + # 'visit DESC,' is optional with Cassandra 4, but ScyllaDB needs it + "WHERE origin = ? AND visit = ? ORDER BY visit DESC, date DESC", ) def origin_visit_status_get( self, origin: str, visit: int, *, statement, @@ -1059,8 +1061,12 @@ @_prepared_select_statement( RawExtrinsicMetadataRow, - "WHERE target=? AND authority_type=? AND authority_url=? " - "AND (discovery_date, id) > (?, ?)", + # This is equivalent to: + # WHERE target=? AND authority_type = ? AND authority_url = ? " + # AND (discovery_date, id) > (?, ?)" + # but it needs to be written this way to work with ScyllaDB. + "WHERE target=? AND (authority_type, authority_url) <= (?, ?) " + "AND (authority_type, authority_url, discovery_date, id) > (?, ?, ?, ?)", ) def raw_extrinsic_metadata_get_after_date_and_id( self, @@ -1076,7 +1082,15 @@ RawExtrinsicMetadataRow.from_dict, self._execute_with_retries( statement, - [target, authority_type, authority_url, after_date, after_id,], + [ + target, + authority_type, + authority_url, + authority_type, + authority_url, + after_date, + after_id, + ], ), ) diff --git a/swh/storage/cassandra/schema.py b/swh/storage/cassandra/schema.py --- a/swh/storage/cassandra/schema.py +++ b/swh/storage/cassandra/schema.py @@ -3,39 +3,56 @@ # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information +import os + +_use_scylla = bool(os.environ.get("SWH_USE_SCYLLADB", "")) + +UDF_LANGUAGE = "lua" if _use_scylla else "java" + +if UDF_LANGUAGE == "java": + # For Cassandra + CREATE_TABLES_QUERIES = [ + """ + CREATE OR REPLACE FUNCTION ascii_bins_count_sfunc ( + state tuple>, -- (nb_none, map) + bin_name ascii + ) + CALLED ON NULL INPUT + RETURNS tuple> + LANGUAGE java AS + $$ + if (bin_name == null) { + state.setInt(0, state.getInt(0) + 1); + } + else { + Map counters = state.getMap( + 1, String.class, Integer.class); + Integer nb = counters.get(bin_name); + if (nb == null) { + nb = 0; + } + counters.put(bin_name, nb + 1); + state.setMap(1, counters, String.class, Integer.class); + } + return state; + $$;""", + """ + CREATE OR REPLACE AGGREGATE ascii_bins_count ( ascii ) + SFUNC ascii_bins_count_sfunc + STYPE tuple> + INITCOND (0, {}) + ;""", + ] +elif UDF_LANGUAGE == "lua": + # For ScyllaDB + # TODO: this is not implementable yet, because ScyllaDB does not support + # user-defined aggregates. https://github.com/scylladb/scylla/issues/7201 + CREATE_TABLES_QUERIES = [] +else: + assert False, f"{UDF_LANGUAGE} must be 'lua' or 'java'" CREATE_TABLES_QUERIES = [ - """ -CREATE OR REPLACE FUNCTION ascii_bins_count_sfunc ( - state tuple>, -- (nb_none, map) - bin_name ascii -) -CALLED ON NULL INPUT -RETURNS tuple> -LANGUAGE java AS -$$ - if (bin_name == null) { - state.setInt(0, state.getInt(0) + 1); - } - else { - Map counters = state.getMap( - 1, String.class, Integer.class); - Integer nb = counters.get(bin_name); - if (nb == null) { - nb = 0; - } - counters.put(bin_name, nb + 1); - state.setMap(1, counters, String.class, Integer.class); - } - return state; -$$ -;""", - """ -CREATE OR REPLACE AGGREGATE ascii_bins_count ( ascii ) -SFUNC ascii_bins_count_sfunc -STYPE tuple> -INITCOND (0, {}) -;""", + *CREATE_TABLES_QUERIES, """ CREATE TYPE IF NOT EXISTS microtimestamp ( seconds bigint, @@ -162,7 +179,9 @@ metadata text, snapshot blob, PRIMARY KEY ((origin), visit, date) -);""", +) +WITH CLUSTERING ORDER BY (visit DESC, date DESC) +;""", # 'WITH CLUSTERING ORDER BY' is optional with Cassandra 4, but ScyllaDB needs it """ CREATE TABLE IF NOT EXISTS origin ( sha1 blob PRIMARY KEY, diff --git a/swh/storage/tests/test_cassandra.py b/swh/storage/tests/test_cassandra.py --- a/swh/storage/tests/test_cassandra.py +++ b/swh/storage/tests/test_cassandra.py @@ -6,6 +6,7 @@ import datetime import itertools import os +import resource import signal import socket import subprocess @@ -58,6 +59,15 @@ commitlog_sync_period_in_ms: 100000 """ +SCYLLA_EXTRA_CONFIG_TEMPLATE = """ +experimental_features: + - udf +view_hints_directory: {data_dir}/view_hints +prometheus_port: 0 # disable prometheus server +start_rpc: false # disable thrift server +api_port: {api_port} +""" + def free_port(): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -68,7 +78,7 @@ def wait_for_peer(addr, port): - wait_until = time.time() + 20 + wait_until = time.time() + 60 while time.time() < wait_until: try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -89,13 +99,29 @@ native_transport_port = free_port() storage_port = free_port() jmx_port = free_port() + api_port = free_port() + + use_scylla = bool(os.environ.get("SWH_USE_SCYLLADB", "")) + + cassandra_bin = os.environ.get( + "SWH_CASSANDRA_BIN", "/usr/bin/scylla" if use_scylla else "/usr/sbin/cassandra" + ) + + if use_scylla: + os.makedirs(cassandra_conf.join("conf")) + config_path = cassandra_conf.join("conf/scylla.yaml") + config_template = CONFIG_TEMPLATE + SCYLLA_EXTRA_CONFIG_TEMPLATE + else: + config_path = cassandra_conf.join("cassandra.yaml") + config_template = CONFIG_TEMPLATE - with open(str(cassandra_conf.join("cassandra.yaml")), "w") as fd: + with open(str(config_path), "w") as fd: fd.write( - CONFIG_TEMPLATE.format( + config_template.format( data_dir=str(cassandra_data), storage_port=storage_port, native_transport_port=native_transport_port, + api_port=api_port, ) ) @@ -104,7 +130,6 @@ else: stdout = stderr = subprocess.DEVNULL - cassandra_bin = os.environ.get("SWH_CASSANDRA_BIN", "/usr/sbin/cassandra") env = { "MAX_HEAP_SIZE": "300M", "HEAP_NEWSIZE": "50M", @@ -113,19 +138,36 @@ if "JAVA_HOME" in os.environ: env["JAVA_HOME"] = os.environ["JAVA_HOME"] - proc = subprocess.Popen( - [ - cassandra_bin, - "-Dcassandra.config=file://%s/cassandra.yaml" % cassandra_conf, - "-Dcassandra.logdir=%s" % cassandra_log, - "-Dcassandra.jmx.local.port=%d" % jmx_port, - "-Dcassandra-foreground=yes", - ], - start_new_session=True, - env=env, - stdout=stdout, - stderr=stderr, - ) + if use_scylla: + env = { + **env, + "SCYLLA_HOME": cassandra_conf, + } + # prevent "NOFILE rlimit too low (recommended setting 200000, + # minimum setting 10000; refusing to start." + resource.setrlimit(resource.RLIMIT_NOFILE, (200000, 200000)) + + proc = subprocess.Popen( + [cassandra_bin, "--developer-mode=1",], + start_new_session=True, + env=env, + stdout=stdout, + stderr=stderr, + ) + else: + proc = subprocess.Popen( + [ + cassandra_bin, + "-Dcassandra.config=file://%s/cassandra.yaml" % cassandra_conf, + "-Dcassandra.logdir=%s" % cassandra_log, + "-Dcassandra.jmx.local.port=%d" % jmx_port, + "-Dcassandra-foreground=yes", + ], + start_new_session=True, + env=env, + stdout=stdout, + stderr=stderr, + ) listening = wait_for_peer("127.0.0.1", native_transport_port)