First implementation of K2V (#293)

**Specification:**

View spec at [this URL](https://git.deuxfleurs.fr/Deuxfleurs/garage/src/branch/k2v/doc/drafts/k2v-spec.md)

- [x] Specify the structure of K2V triples
- [x] Specify the DVVS format used for causality detection
- [x] Specify the K2V index (just a counter of number of values per partition key)
- [x] Specify single-item endpoints: ReadItem, InsertItem, DeleteItem
- [x] Specify index endpoint: ReadIndex
- [x] Specify multi-item endpoints: InsertBatch, ReadBatch, DeleteBatch
- [x] Move to JSON objects instead of tuples
- [x] Specify endpoints for polling for updates on single values (PollItem)

**Implementation:**

- [x] Table for K2V items, causal contexts
- [x] Indexing mechanism and table for K2V index
- [x] Make API handlers a bit more generic
- [x] K2V API endpoint
- [x] K2V API router
- [x] ReadItem
- [x] InsertItem
- [x] DeleteItem
- [x] PollItem
- [x] ReadIndex
- [x] InsertBatch
- [x] ReadBatch
- [x] DeleteBatch

**Testing:**

- [x] Just a simple Python script that does some requests to check visually that things are going right (does not contain parsing of results or assertions on returned values)
- [x] Actual tests:
  - [x] Adapt testing framework
  - [x] Simple test with InsertItem + ReadItem
  - [x] Test with several Insert/Read/DeleteItem + ReadIndex
  - [x] Test all combinations of return formats for ReadItem
  - [x] Test with ReadBatch, InsertBatch, DeleteBatch
  - [x] Test with PollItem
  - [x] Test error codes
- [ ] Fix most broken stuff
  - [x] test PollItem broken randomly
  - [x] when invalid causality tokens are given, errors should be 4xx not 5xx

**Improvements:**

- [x] Descending range queries
  - [x] Specify
  - [x] Implement
  - [x] Add test
- [x] Batch updates to index counter
- [x] Put K2V behind `k2v` feature flag

Co-authored-by: Alex Auvolat <alex@adnab.me>
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/293
Co-authored-by: Alex <alex@adnab.me>
Co-committed-by: Alex <alex@adnab.me>

1 files changed, 252 insertions, 0 deletions

diff --git a/src/api/k2v/router.rs b/src/api/k2v/router.rs
new file mode 100644
index 00000000..f948ffce
--- /dev/null
+++ b/src/api/k2v/router.rs
@@ -0,0 +1,252 @@
+use crate::error::*;
+
+use std::borrow::Cow;
+
+use hyper::{Method, Request};
+
+use crate::helpers::Authorization;
+use crate::router_macros::{generateQueryParameters, router_match};
+
+router_match! {@func
+
+
+/// List of all K2V API endpoints.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Endpoint {
+	DeleteBatch {
+	},
+	DeleteItem {
+		partition_key: String,
+		sort_key: String,
+	},
+	InsertBatch {
+	},
+	InsertItem {
+		partition_key: String,
+		sort_key: String,
+	},
+	Options,
+	PollItem {
+		partition_key: String,
+		sort_key: String,
+		causality_token: String,
+		timeout: Option<u64>,
+	},
+	ReadBatch {
+	},
+	ReadIndex {
+		prefix: Option<String>,
+		start: Option<String>,
+		end: Option<String>,
+		limit: Option<u64>,
+		reverse: Option<bool>,
+	},
+	ReadItem {
+		partition_key: String,
+		sort_key: String,
+	},
+}}
+
+impl Endpoint {
+	/// Determine which S3 endpoint a request is for using the request, and a bucket which was
+	/// possibly extracted from the Host header.
+	/// Returns Self plus bucket name, if endpoint is not Endpoint::ListBuckets
+	pub fn from_request<T>(req: &Request<T>) -> Result<(Self, String), Error> {
+		let uri = req.uri();
+		let path = uri.path().trim_start_matches('/');
+		let query = uri.query();
+
+		let (bucket, partition_key) = path
+			.split_once('/')
+			.map(|(b, p)| (b.to_owned(), p.trim_start_matches('/')))
+			.unwrap_or((path.to_owned(), ""));
+
+		if bucket.is_empty() {
+			return Err(Error::BadRequest("Missing bucket name".to_owned()));
+		}
+
+		if *req.method() == Method::OPTIONS {
+			return Ok((Self::Options, bucket));
+		}
+
+		let partition_key = percent_encoding::percent_decode_str(partition_key)
+			.decode_utf8()?
+			.into_owned();
+
+		let mut query = QueryParameters::from_query(query.unwrap_or_default())?;
+
+		let method_search = Method::from_bytes(b"SEARCH").unwrap();
+		let res = match *req.method() {
+			Method::GET => Self::from_get(partition_key, &mut query)?,
+			//&Method::HEAD => Self::from_head(partition_key, &mut query)?,
+			Method::POST => Self::from_post(partition_key, &mut query)?,
+			Method::PUT => Self::from_put(partition_key, &mut query)?,
+			Method::DELETE => Self::from_delete(partition_key, &mut query)?,
+			_ if req.method() == method_search => Self::from_search(partition_key, &mut query)?,
+			_ => return Err(Error::BadRequest("Unknown method".to_owned())),
+		};
+
+		if let Some(message) = query.nonempty_message() {
+			debug!("Unused query parameter: {}", message)
+		}
+		Ok((res, bucket))
+	}
+
+	/// Determine which endpoint a request is for, knowing it is a GET.
+	fn from_get(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+				EMPTY if causality_token => PollItem (query::sort_key, query::causality_token, opt_parse::timeout),
+				EMPTY => ReadItem (query::sort_key),
+			],
+			no_key: [
+				EMPTY => ReadIndex (query_opt::prefix, query_opt::start, query_opt::end, opt_parse::limit, opt_parse::reverse),
+			]
+		}
+	}
+
+	/// Determine which endpoint a request is for, knowing it is a SEARCH.
+	fn from_search(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+			],
+			no_key: [
+				EMPTY => ReadBatch,
+			]
+		}
+	}
+
+	/*
+	/// Determine which endpoint a request is for, knowing it is a HEAD.
+	fn from_head(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+				EMPTY => HeadObject(opt_parse::part_number, query_opt::version_id),
+			],
+			no_key: [
+				EMPTY => HeadBucket,
+			]
+		}
+	}
+	*/
+
+	/// Determine which endpoint a request is for, knowing it is a POST.
+	fn from_post(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+			],
+			no_key: [
+				EMPTY => InsertBatch,
+				DELETE => DeleteBatch,
+				SEARCH => ReadBatch,
+			]
+		}
+	}
+
+	/// Determine which endpoint a request is for, knowing it is a PUT.
+	fn from_put(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+				EMPTY => InsertItem (query::sort_key),
+
+			],
+			no_key: [
+			]
+		}
+	}
+
+	/// Determine which endpoint a request is for, knowing it is a DELETE.
+	fn from_delete(partition_key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
+		router_match! {
+			@gen_parser
+			(query.keyword.take().unwrap_or_default().as_ref(), partition_key, query, None),
+			key: [
+				EMPTY => DeleteItem (query::sort_key),
+			],
+			no_key: [
+			]
+		}
+	}
+
+	/// Get the partition key the request target. Returns None for requests which don't use a partition key.
+	#[allow(dead_code)]
+	pub fn get_partition_key(&self) -> Option<&str> {
+		router_match! {
+			@extract
+			self,
+			partition_key,
+			[
+				DeleteItem,
+				InsertItem,
+				PollItem,
+				ReadItem,
+			]
+		}
+	}
+
+	/// Get the sort key the request target. Returns None for requests which don't use a sort key.
+	#[allow(dead_code)]
+	pub fn get_sort_key(&self) -> Option<&str> {
+		router_match! {
+			@extract
+			self,
+			sort_key,
+			[
+				DeleteItem,
+				InsertItem,
+				PollItem,
+				ReadItem,
+			]
+		}
+	}
+
+	/// Get the kind of authorization which is required to perform the operation.
+	pub fn authorization_type(&self) -> Authorization {
+		let readonly = router_match! {
+			@match
+			self,
+			[
+				PollItem,
+				ReadBatch,
+				ReadIndex,
+				ReadItem,
+			]
+		};
+		if readonly {
+			Authorization::Read
+		} else {
+			Authorization::Write
+		}
+	}
+}
+
+// parameter name => struct field
+generateQueryParameters! {
+	"prefix" => prefix,
+	"start" => start,
+	"causality_token" => causality_token,
+	"end" => end,
+	"limit" => limit,
+	"reverse" => reverse,
+	"sort_key" => sort_key,
+	"timeout" => timeout
+}
+
+mod keywords {
+	//! This module contain all query parameters with no associated value
+	//! used to differentiate endpoints.
+	pub const EMPTY: &str = "";
+
+	pub const DELETE: &str = "delete";
+	pub const SEARCH: &str = "search";
+}