Add unixctl library to interface with OVS

.github/workflows/ci.yaml

@@ -0,0 +1,53 @@
+name: CI
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    container: quay.io/centos/centos:stream9
+    env:
+      CARGO_HOME: ${{ github.workspace}}/.cargo
+      RUSTUP_HOME: ${{ github.workspace}}/.rustup
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Install NFV SIG repository
+        run: |
+          dnf install -y centos-release-nfv-openvswitch
+
+      - name: Install needed packages
+        run: |
+          dnf install -y gcc openvswitch3.1
+
+      - name: Install Rustup
+        run: |
+          curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
+          source $CARGO_HOME/env
+
+      - name: Add Cargo to PATH
+        run: echo "$GITHUB_WORKSPACE/.cargo/bin" >> $GITHUB_PATH
+
+      - name: Install Rust toolchain
+        run: rustup install stable
+
+      - name: Install Clippy
+        run: rustup component add clippy
+
+      - name: Run Linter
+        run: cargo fmt --check
+
+      - name: Run Clippy
+        run: cargo clippy --all-targets --all-features -- -D warnings
+
+      - name: Run Integration Tests
+        run: cargo test -F test_integration
+

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "ovs-unixctl"
+version = "0.1.0"
+license = "GPL-2.0-only"
+description = "Control OVS daemons using the Unixctl interface (a.k.a ovs-appctl)"
+readme = "README.md"
+edition = "2021"
+
+[features]
+test_integration = []
+
+[dependencies]
+anyhow = "1.0"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0", features = ["raw_value"] }

README.md

@@ -1,2 +1,17 @@
 # ovs-unixctl
-Control OVS daemons using the Unixctl interface (a.k.a ovs-appctl)
+Library to send commands to OVS daemons though their JSON interface.
+See **ovs-appctl(8)**.
+
+## Test
+
+Run unit tests:
+
+```
+$ cargo test
+```
+
+Run integration tests, if openvswitch is installed in the system:
+
+```
+$ cargo test -F test_integration
+```

src/jsonrpc.rs

@@ -0,0 +1,154 @@
+//! A simple JSON-RPC client compatible with OVS unixctl.
+
+use std::{
+    fmt, path,
+    sync::atomic::{AtomicUsize, Ordering::Relaxed},
+    time,
+};
+
+use anyhow::{anyhow, bail, Result};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+
+use crate::unix;
+
+// JsonStreams are capable of sending and receiving JSON messages.
+pub(crate) trait JsonStream: Send + Sync + 'static {
+    /// Send a message to the target.
+    fn send<M: Serialize>(&mut self, msg: M) -> Result<()>;
+
+    /// Receivea message from the target (blocking).
+    fn recv<R>(&mut self) -> Result<R>
+    where
+        R: for<'a> Deserialize<'a>;
+}
+
+// Client streams can connect and disconnect from targets creating
+// some JsonStream.
+pub(crate) trait JsonStreamClient: fmt::Display {
+    type Stream: JsonStream;
+    /// Connect to the target.
+    fn connect(&mut self) -> Result<Self::Stream>;
+}
+
+/// A JSON-RPC request.
+#[derive(Debug, Serialize)]
+pub struct Request<'a, P: Serialize + AsRef<str> = &'a str> {
+    /// The name of the RPC call.
+    pub method: &'a str,
+    /// Parameters to the RPC call.
+    pub params: &'a [P],
+    /// Identifier for this request, which should appear in the response.
+    pub id: usize,
+}
+
+/// A JSONRPC response object.
+/// TODO make generic
+#[derive(Debug, Clone, Deserialize)]
+pub(crate) struct Response<R = String> {
+    /// The result of the request.
+    pub result: Option<R>,
+    /// An error if it occurred.
+    pub error: Option<String>,
+    /// Identifier for this response. It should match that of the associated request.
+    pub id: Option<usize>,
+}
+
+/// JSON-RPC client.
+#[derive(Debug)]
+pub(crate) struct Client<C: JsonStreamClient> {
+    stream_client: C,
+    stream: Option<C::Stream>,
+    last_id: AtomicUsize,
+}
+
+impl<C: JsonStreamClient> Client<C> {
+    /// Creates a new client with the given transport.
+    pub(crate) fn new(stream_client: C) -> Client<C> {
+        Client {
+            stream_client,
+            stream: None,
+            last_id: AtomicUsize::new(1),
+        }
+    }
+
+    /// Creates a new client with a Unix socket transport.
+    pub(crate) fn unix<P: AsRef<path::Path>>(
+        sock_path: P,
+        timeout: Option<time::Duration>,
+    ) -> Client<unix::UnixJsonStreamClient> {
+        let mut stream_client = unix::UnixJsonStreamClient::new(sock_path);
+        if let Some(timeout) = timeout {
+            stream_client = stream_client.timeout(timeout);
+        }
+        Client::new(stream_client)
+    }
+
+    /// Builds a request with the given method and parameters.
+    ///
+    /// It internally deals with incrementing the id.
+    fn build_request<'a, P: Serialize + AsRef<str>>(
+        &self,
+        method: &'a str,
+        params: &'a [P],
+    ) -> Request<'a, P> {
+        Request {
+            method,
+            params,
+            id: self.last_id.fetch_add(1, Relaxed),
+        }
+    }
+
+    /// Sends a request and returns the response.
+    pub fn send_request<R: DeserializeOwned, P: Serialize + AsRef<str>>(
+        &mut self,
+        request: Request<P>,
+    ) -> Result<Response<R>> {
+        if self.stream.is_none() {
+            self.stream = Some(self.stream_client.connect()?);
+        }
+
+        let stream = self.stream.as_mut().unwrap();
+        let req_id = request.id;
+
+        stream.send(request)?;
+        let res: Response<R> = stream.recv()?;
+        if res.id.ok_or_else(|| anyhow!("no id present in response"))? != req_id {
+            bail!("ID missmatch");
+        }
+
+        Ok(res)
+    }
+
+    /// Calls a method with some arguments and returns the result.
+    pub(crate) fn call_params<R: DeserializeOwned, P: Serialize + AsRef<str>>(
+        &mut self,
+        method: &str,
+        params: &[P],
+    ) -> Result<Response<R>> {
+        let request = self.build_request(method, params);
+        let response = self.send_request(request)?;
+        if let Some(error) = response.error {
+            bail!(
+                "Failed to run command {} with params [{}]: {}",
+                method,
+                params
+                    .iter()
+                    .map(|p| p.as_ref())
+                    .collect::<Vec<&str>>()
+                    .join(", "),
+                error,
+            )
+        }
+        Ok(response)
+    }
+
+    /// Calls a method without arguments and resturns the result.
+    pub(crate) fn call<R: DeserializeOwned>(&mut self, method: &str) -> Result<Response<R>> {
+        let request = self.build_request::<&str>(method, &[]);
+        let response = self.send_request(request)?;
+        if let Some(error) = response.error {
+            bail!("Failed to run command {}: {}", method, error,)
+        }
+        Ok(response)
+    }
+}

src/lib.rs

@@ -0,0 +1,8 @@
+//! OpenvSwitch application control (appctl) library.
+
+//FIXME
+#[allow(dead_code)]
+pub mod jsonrpc;
+pub mod ovs;
+pub use ovs::*;
+pub mod unix;