You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/02/17 10:42:23 UTC
(opendal) branch ghac-v2 created (now 3cdfa1a5a)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch ghac-v2
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 3cdfa1a5a test againest ghac v2
This branch includes the following new commits:
new b6c362854 refactor ghac into core based
new de4ab91cf Implement ghac v2 support
new 3cdfa1a5a test againest ghac v2
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/03: refactor ghac into core based
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch ghac-v2
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit b6c3628546deff6d298766adaa201f73d5ca8397
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Feb 17 15:40:36 2025 +0800
refactor ghac into core based
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/ghac/backend.rs | 192 +++++--------------------------------
core/src/services/ghac/core.rs | 194 ++++++++++++++++++++++++++++++++++++++
core/src/services/ghac/mod.rs | 2 +
core/src/services/ghac/writer.rs | 17 ++--
4 files changed, 227 insertions(+), 178 deletions(-)
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index 423bde57e..64361c198 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -19,40 +19,20 @@ use std::env;
use std::sync::Arc;
use bytes::Buf;
-use bytes::Bytes;
use http::header;
-use http::header::ACCEPT;
-use http::header::AUTHORIZATION;
-use http::header::CONTENT_LENGTH;
-use http::header::CONTENT_RANGE;
-use http::header::CONTENT_TYPE;
use http::Request;
use http::Response;
use http::StatusCode;
use log::debug;
-use serde::Deserialize;
-use serde::Serialize;
+use super::core::*;
use super::error::parse_error;
use super::writer::GhacWriter;
use crate::raw::*;
+use crate::services::ghac::core::GhacCore;
use crate::services::GhacConfig;
use crate::*;
-/// The base url for cache url.
-const CACHE_URL_BASE: &str = "_apis/artifactcache";
-/// Cache API requires to provide an accept header.
-const CACHE_HEADER_ACCEPT: &str = "application/json;api-version=6.0-preview.1";
-/// The cache url env for ghac.
-///
-/// The url will be like `https://artifactcache.actions.githubusercontent.com/<id>/`
-const ACTIONS_CACHE_URL: &str = "ACTIONS_CACHE_URL";
-/// The runtime token env for ghac.
-///
-/// This token will be valid for 6h and github action will running for 6
-/// hours at most. So we don't need to refetch it again.
-const ACTIONS_RUNTIME_TOKEN: &str = "ACTIONS_RUNTIME_TOKEN";
-
fn value_or_env(
explicit_value: Option<String>,
env_var_name: &str,
@@ -173,7 +153,7 @@ impl Builder for GhacBuilder {
})?
};
- let backend = GhacBackend {
+ let core = GhacCore {
root,
cache_url: value_or_env(self.config.endpoint, ACTIONS_CACHE_URL, "Builder::build")?,
@@ -191,21 +171,16 @@ impl Builder for GhacBuilder {
client,
};
- Ok(backend)
+ Ok(GhacBackend {
+ core: Arc::new(core),
+ })
}
}
/// Backend for github action cache services.
#[derive(Debug, Clone)]
pub struct GhacBackend {
- // root should end with "/"
- root: String,
-
- cache_url: String,
- catch_token: String,
- version: String,
-
- pub client: HttpClient,
+ core: Arc<GhacCore>,
}
impl Access for GhacBackend {
@@ -221,8 +196,8 @@ impl Access for GhacBackend {
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
am.set_scheme(Scheme::Ghac)
- .set_root(&self.root)
- .set_name(&self.version)
+ .set_root(&self.core.root)
+ .set_name(&self.core.version)
.set_native_capability(Capability {
stat: true,
stat_has_cache_control: true,
@@ -253,9 +228,9 @@ impl Access for GhacBackend {
///
/// In this way, we can support both self-hosted GHES and `github.com`.
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let req = self.ghac_query(path)?;
+ let req = self.core.ghac_query(path)?;
- let resp = self.client.send(req).await?;
+ let resp = self.core.client.send(req).await?;
let location = if resp.status() == StatusCode::OK {
let slc = resp.into_body();
@@ -270,7 +245,7 @@ impl Access for GhacBackend {
.header(header::RANGE, "bytes=0-0")
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = self.client.send(req).await?;
+ let resp = self.core.client.send(req).await?;
let status = resp.status();
match status {
@@ -291,9 +266,9 @@ impl Access for GhacBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let req = self.ghac_query(path)?;
+ let req = self.core.ghac_query(path)?;
- let resp = self.client.send(req).await?;
+ let resp = self.core.client.send(req).await?;
let location = if resp.status() == StatusCode::OK {
let slc = resp.into_body();
@@ -304,8 +279,8 @@ impl Access for GhacBackend {
return Err(parse_error(resp));
};
- let req = self.ghac_get_location(&location, args.range())?;
- let resp = self.client.fetch(req).await?;
+ let req = self.core.ghac_get_location(&location, args.range())?;
+ let resp = self.core.client.fetch(req).await?;
let status = resp.status();
match status {
@@ -321,9 +296,9 @@ impl Access for GhacBackend {
}
async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let req = self.ghac_reserve(path)?;
+ let req = self.core.ghac_reserve(path)?;
- let resp = self.client.send(req).await?;
+ let resp = self.core.client.send(req).await?;
let cache_id = if resp.status().is_success() {
let slc = resp.into_body();
@@ -334,132 +309,9 @@ impl Access for GhacBackend {
return Err(parse_error(resp).map(|err| err.with_operation("Backend::ghac_reserve")));
};
- Ok((RpWrite::default(), GhacWriter::new(self.clone(), cache_id)))
- }
-}
-
-impl GhacBackend {
- fn ghac_query(&self, path: &str) -> Result<Request<Buffer>> {
- let p = build_abs_path(&self.root, path);
-
- let url = format!(
- "{}{CACHE_URL_BASE}/cache?keys={}&version={}",
- self.cache_url,
- percent_encode_path(&p),
- self.version
- );
-
- let mut req = Request::get(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
-
- let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
-
- Ok(req)
- }
-
- pub fn ghac_get_location(&self, location: &str, range: BytesRange) -> Result<Request<Buffer>> {
- let mut req = Request::get(location);
-
- if !range.is_full() {
- req = req.header(header::RANGE, range.to_header());
- }
-
- req.body(Buffer::new()).map_err(new_request_build_error)
- }
-
- fn ghac_reserve(&self, path: &str) -> Result<Request<Buffer>> {
- let p = build_abs_path(&self.root, path);
-
- let url = format!("{}{CACHE_URL_BASE}/caches", self.cache_url);
-
- let bs = serde_json::to_vec(&GhacReserveRequest {
- key: p,
- version: self.version.to_string(),
- })
- .map_err(new_json_serialize_error)?;
-
- let mut req = Request::post(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_LENGTH, bs.len());
- req = req.header(CONTENT_TYPE, "application/json");
-
- let req = req
- .body(Buffer::from(Bytes::from(bs)))
- .map_err(new_request_build_error)?;
-
- Ok(req)
- }
-
- pub fn ghac_upload(
- &self,
- cache_id: i64,
- offset: u64,
- size: u64,
- body: Buffer,
- ) -> Result<Request<Buffer>> {
- let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
-
- let mut req = Request::patch(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_LENGTH, size);
- req = req.header(CONTENT_TYPE, "application/octet-stream");
- req = req.header(
- CONTENT_RANGE,
- BytesContentRange::default()
- .with_range(offset, offset + size - 1)
- .to_header(),
- );
-
- let req = req.body(body).map_err(new_request_build_error)?;
-
- Ok(req)
+ Ok((
+ RpWrite::default(),
+ GhacWriter::new(self.core.clone(), cache_id),
+ ))
}
-
- pub fn ghac_commit(&self, cache_id: i64, size: u64) -> Result<Request<Buffer>> {
- let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
-
- let bs =
- serde_json::to_vec(&GhacCommitRequest { size }).map_err(new_json_serialize_error)?;
-
- let mut req = Request::post(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_TYPE, "application/json");
- req = req.header(CONTENT_LENGTH, bs.len());
-
- let req = req
- .body(Buffer::from(Bytes::from(bs)))
- .map_err(new_request_build_error)?;
-
- Ok(req)
- }
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct GhacQueryResponse {
- // Not used fields.
- // cache_key: String,
- // scope: String,
- archive_location: String,
-}
-
-#[derive(Serialize)]
-struct GhacReserveRequest {
- key: String,
- version: String,
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct GhacReserveResponse {
- cache_id: i64,
-}
-
-#[derive(Serialize)]
-struct GhacCommitRequest {
- size: u64,
}
diff --git a/core/src/services/ghac/core.rs b/core/src/services/ghac/core.rs
new file mode 100644
index 000000000..a9c6d5eb7
--- /dev/null
+++ b/core/src/services/ghac/core.rs
@@ -0,0 +1,194 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use crate::raw::{
+ build_abs_path, new_json_serialize_error, new_request_build_error, percent_encode_path,
+ BytesContentRange, BytesRange, HttpClient,
+};
+use crate::Buffer;
+use bytes::Bytes;
+use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE};
+use http::{header, Request};
+use serde::{Deserialize, Serialize};
+use std::fmt::{Debug, Formatter};
+
+/// The base url for cache url.
+pub const CACHE_URL_BASE: &str = "_apis/artifactcache";
+/// Cache API requires to provide an accept header.
+pub const CACHE_HEADER_ACCEPT: &str = "application/json;api-version=6.0-preview.1";
+/// The cache url env for ghac.
+///
+/// The url will be like `https://artifactcache.actions.githubusercontent.com/<id>/`
+pub const ACTIONS_CACHE_URL: &str = "ACTIONS_CACHE_URL";
+/// The runtime token env for ghac.
+///
+/// This token will be valid for 6h and github action will running for 6
+/// hours at most. So we don't need to refetch it again.
+pub const ACTIONS_RUNTIME_TOKEN: &str = "ACTIONS_RUNTIME_TOKEN";
+
+/// Core for github action cache services.
+#[derive(Clone)]
+pub struct GhacCore {
+ // root should end with "/"
+ pub root: String,
+
+ pub cache_url: String,
+ pub catch_token: String,
+ pub version: String,
+
+ pub client: HttpClient,
+}
+
+impl Debug for GhacCore {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("GhacCore")
+ .field("root", &self.root)
+ .field("cache_url", &self.cache_url)
+ .field("version", &self.version)
+ .finish_non_exhaustive()
+ }
+}
+
+impl GhacCore {
+ pub fn ghac_query(&self, path: &str) -> crate::Result<Request<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "{}{CACHE_URL_BASE}/cache?keys={}&version={}",
+ self.cache_url,
+ percent_encode_path(&p),
+ self.version
+ );
+
+ let mut req = Request::get(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+
+ let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+
+ pub fn ghac_get_location(
+ &self,
+ location: &str,
+ range: BytesRange,
+ ) -> crate::Result<Request<Buffer>> {
+ let mut req = Request::get(location);
+
+ if !range.is_full() {
+ req = req.header(header::RANGE, range.to_header());
+ }
+
+ req.body(Buffer::new()).map_err(new_request_build_error)
+ }
+
+ pub fn ghac_reserve(&self, path: &str) -> crate::Result<Request<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+
+ let url = format!("{}{CACHE_URL_BASE}/caches", self.cache_url);
+
+ let bs = serde_json::to_vec(&GhacReserveRequest {
+ key: p,
+ version: self.version.to_string(),
+ })
+ .map_err(new_json_serialize_error)?;
+
+ let mut req = Request::post(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+ req = req.header(CONTENT_LENGTH, bs.len());
+ req = req.header(CONTENT_TYPE, "application/json");
+
+ let req = req
+ .body(Buffer::from(Bytes::from(bs)))
+ .map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+
+ pub fn ghac_upload(
+ &self,
+ cache_id: i64,
+ offset: u64,
+ size: u64,
+ body: Buffer,
+ ) -> crate::Result<Request<Buffer>> {
+ let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
+
+ let mut req = Request::patch(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+ req = req.header(CONTENT_LENGTH, size);
+ req = req.header(CONTENT_TYPE, "application/octet-stream");
+ req = req.header(
+ CONTENT_RANGE,
+ BytesContentRange::default()
+ .with_range(offset, offset + size - 1)
+ .to_header(),
+ );
+
+ let req = req.body(body).map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+
+ pub fn ghac_commit(&self, cache_id: i64, size: u64) -> crate::Result<Request<Buffer>> {
+ let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
+
+ let bs =
+ serde_json::to_vec(&GhacCommitRequest { size }).map_err(new_json_serialize_error)?;
+
+ let mut req = Request::post(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+ req = req.header(CONTENT_TYPE, "application/json");
+ req = req.header(CONTENT_LENGTH, bs.len());
+
+ let req = req
+ .body(Buffer::from(Bytes::from(bs)))
+ .map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GhacQueryResponse {
+ // Not used fields.
+ // cache_key: String,
+ // scope: String,
+ pub archive_location: String,
+}
+
+#[derive(Serialize)]
+pub struct GhacReserveRequest {
+ pub key: String,
+ pub version: String,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GhacReserveResponse {
+ pub cache_id: i64,
+}
+
+#[derive(Serialize)]
+pub struct GhacCommitRequest {
+ pub size: u64,
+}
diff --git a/core/src/services/ghac/mod.rs b/core/src/services/ghac/mod.rs
index fabc5bcc1..992ecb2b6 100644
--- a/core/src/services/ghac/mod.rs
+++ b/core/src/services/ghac/mod.rs
@@ -26,4 +26,6 @@ mod backend;
pub use backend::GhacBuilder as Ghac;
mod config;
+mod core;
+
pub use config::GhacConfig;
diff --git a/core/src/services/ghac/writer.rs b/core/src/services/ghac/writer.rs
index c5d49a17a..90b39c564 100644
--- a/core/src/services/ghac/writer.rs
+++ b/core/src/services/ghac/writer.rs
@@ -15,22 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-use super::backend::GhacBackend;
+use super::core::*;
use super::error::parse_error;
use crate::raw::*;
use crate::*;
+use std::sync::Arc;
pub struct GhacWriter {
- backend: GhacBackend,
+ core: Arc<GhacCore>,
cache_id: i64,
size: u64,
}
impl GhacWriter {
- pub fn new(backend: GhacBackend, cache_id: i64) -> Self {
+ pub fn new(core: Arc<GhacCore>, cache_id: i64) -> Self {
GhacWriter {
- backend,
+ core,
cache_id,
size: 0,
}
@@ -42,14 +43,14 @@ impl oio::Write for GhacWriter {
let size = bs.len();
let offset = self.size;
- let req = self.backend.ghac_upload(
+ let req = self.core.ghac_upload(
self.cache_id,
offset,
size as u64,
Buffer::from(bs.to_bytes()),
)?;
- let resp = self.backend.client.send(req).await?;
+ let resp = self.core.client.send(req).await?;
if !resp.status().is_success() {
return Err(parse_error(resp).map(|err| err.with_operation("Backend::ghac_upload")));
@@ -64,8 +65,8 @@ impl oio::Write for GhacWriter {
}
async fn close(&mut self) -> Result<()> {
- let req = self.backend.ghac_commit(self.cache_id, self.size)?;
- let resp = self.backend.client.send(req).await?;
+ let req = self.core.ghac_commit(self.cache_id, self.size)?;
+ let resp = self.core.client.send(req).await?;
if resp.status().is_success() {
Ok(())
(opendal) 03/03: test againest ghac v2
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch ghac-v2
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 3cdfa1a5a2556080d600b8f8b9c5501c5286db73
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Feb 17 18:42:05 2025 +0800
test againest ghac v2
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
.github/workflows/service_test_ghac.yml | 31 +++++++++++++++-
core/src/services/ghac/backend.rs | 46 +++++++++++++++++++-----
core/src/services/ghac/core.rs | 64 ++++++++++++++++++++++++++++++++-
3 files changed, 131 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/service_test_ghac.yml b/.github/workflows/service_test_ghac.yml
index 0fcd067bf..e3c2221dc 100644
--- a/.github/workflows/service_test_ghac.yml
+++ b/.github/workflows/service_test_ghac.yml
@@ -37,7 +37,7 @@ concurrency:
cancel-in-progress: true
jobs:
- ghac:
+ ghac_v1:
runs-on: ubuntu-latest
if: github.event_name == 'push' || !github.event.pull_request.head.repo.fork
permissions:
@@ -55,6 +55,35 @@ jobs:
script: |
core.exportVariable('ACTIONS_CACHE_URL', process.env.ACTIONS_CACHE_URL || '');
core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || '');
+ core.exportVariable('ACTIONS_CACHE_SERVICE_V2', '');
+
+ - name: Test
+ shell: bash
+ working-directory: core
+ run: cargo test behavior --features tests,services-ghac
+ env:
+ OPENDAL_TEST: ghac
+ GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+ ghac_v2:
+ runs-on: ubuntu-latest
+ if: github.event_name == 'push' || !github.event.pull_request.head.repo.fork
+ permissions:
+ actions: write
+ steps:
+ - uses: actions/checkout@v4
+ - name: Setup Rust toolchain
+ uses: ./.github/actions/setup
+ with:
+ need-nextest: true
+
+ - name: Configure Cache Env
+ uses: actions/github-script@v7
+ with:
+ script: |
+ core.exportVariable('ACTIONS_RESULTS_URL', process.env.ACTIONS_RESULTS_URL || '');
+ core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || '');
+ core.exportVariable('ACTIONS_CACHE_SERVICE_V2', 'on');
- name: Test
shell: bash
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index f3486f2dd..df50e5bd2 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -16,14 +16,9 @@
// under the License.
use std::env;
+use std::str::FromStr;
use std::sync::Arc;
-use http::header;
-use http::Request;
-use http::Response;
-use http::StatusCode;
-use log::debug;
-
use super::core::*;
use super::error::parse_error;
use super::writer::GhacWriter;
@@ -31,6 +26,13 @@ use crate::raw::*;
use crate::services::ghac::core::GhacCore;
use crate::services::GhacConfig;
use crate::*;
+use ::ghac::v1 as ghac_grpc;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use log::debug;
+use tonic::transport::Channel;
fn value_or_env(
explicit_value: Option<String>,
@@ -143,6 +145,34 @@ impl Builder for GhacBuilder {
let root = normalize_root(&self.config.root.unwrap_or_default());
debug!("backend use root {}", root);
+ let version = get_cache_service_version();
+
+ let cache_url = self
+ .config
+ .endpoint
+ .unwrap_or_else(|| get_cache_service_url(&version));
+ if cache_url.is_empty() {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "cache url for ghac not found, maybe not in github action environment?".to_string(),
+ ));
+ }
+
+ let mut grpc_client = None;
+ // Try connect to grpc endpoint if we are using v2.
+ if version == "v2" {
+ let uri = http::Uri::from_str(&cache_url).map_err(|err| {
+ Error::new(ErrorKind::ConfigInvalid, "invalid cache url for ghac")
+ .with_operation("Builder::build")
+ .set_source(err)
+ })?;
+ let endpoint = Channel::builder(uri);
+ let channel = endpoint.connect_lazy();
+ grpc_client = Some(ghac_grpc::cache_service_client::CacheServiceClient::new(
+ channel,
+ ));
+ }
+
let http_client = if let Some(client) = self.http_client {
client
} else {
@@ -155,7 +185,7 @@ impl Builder for GhacBuilder {
let core = GhacCore {
root,
- cache_url: value_or_env(self.config.endpoint, ACTIONS_CACHE_URL, "Builder::build")?,
+ cache_url,
catch_token: value_or_env(
self.config.runtime_token,
ACTIONS_RUNTIME_TOKEN,
@@ -168,7 +198,7 @@ impl Builder for GhacBuilder {
.unwrap_or_else(|| "opendal".to_string()),
http_client,
- grpc_client: None,
+ grpc_client,
};
Ok(GhacBackend {
diff --git a/core/src/services/ghac/core.rs b/core/src/services/ghac/core.rs
index ea864f768..bc30efec7 100644
--- a/core/src/services/ghac/core.rs
+++ b/core/src/services/ghac/core.rs
@@ -24,9 +24,11 @@ use crate::*;
use ::ghac::v1 as ghac_grpc;
use bytes::{Buf, Bytes};
use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
-use http::{Request, StatusCode};
+use http::{Request, StatusCode, Uri};
use serde::{Deserialize, Serialize};
+use std::env;
use std::fmt::{Debug, Formatter};
+use std::str::FromStr;
use tonic::transport::Channel;
use tonic::IntoRequest;
@@ -43,6 +45,10 @@ pub const ACTIONS_CACHE_URL: &str = "ACTIONS_CACHE_URL";
/// This token will be valid for 6h and github action will running for 6
/// hours at most. So we don't need to refetch it again.
pub const ACTIONS_RUNTIME_TOKEN: &str = "ACTIONS_RUNTIME_TOKEN";
+/// The cache service version env for ghac.
+pub const ACTIONS_CACHE_SERVICE_V2: &str = "ACTIONS_CACHE_SERVICE_V2";
+/// The results url env for ghac.
+pub const ACTIONS_RESULTS_URL: &str = "ACTIONS_RESULTS_URL";
/// Core for github action cache services.
#[derive(Clone)]
@@ -216,6 +222,62 @@ impl GhacCore {
}
}
+/// Determines if the current environment is GitHub Enterprise Server (GHES)
+///
+/// We need to know this since GHES doesn't support ghac v2 yet.
+pub fn is_ghes() -> bool {
+ // Fetch GitHub Server URL with fallback to "https://github.com"
+ let server_url =
+ env::var("GITHUB_SERVER_URL").unwrap_or_else(|_| "https://github.com".to_string());
+
+ let Ok(url) = Uri::from_str(&server_url) else {
+ // We just return false if the URL is invalid
+ return false;
+ };
+
+ // Check against known non-GHES host patterns
+ let hostname = url.host().unwrap_or("").trim_end().to_lowercase();
+
+ let is_github_host = hostname == "github.com";
+ let is_ghe_host = hostname.ends_with(".ghe.com");
+ let is_localhost = hostname.ends_with(".localhost");
+
+ !is_github_host && !is_ghe_host && !is_localhost
+}
+
+/// Determines the cache service version based on environment
+pub fn get_cache_service_version() -> String {
+ if is_ghes() {
+ // GHES only supports v1 regardless of feature flags
+ "v1".to_string()
+ } else {
+ // Check for presence of non-empty ACTIONS_CACHE_SERVICE_V2
+ let value = env::var(ACTIONS_CACHE_SERVICE_V2).unwrap_or_default();
+ if value.is_empty() {
+ "v1".to_string()
+ } else {
+ "v2".to_string()
+ }
+ }
+}
+
+/// Returns the appropriate cache service URL based on version
+pub fn get_cache_service_url(version: &str) -> String {
+ match version {
+ "v1" => {
+ // Priority order for v1: CACHE_URL -> RESULTS_URL
+ env::var(ACTIONS_CACHE_URL)
+ .or_else(|_| env::var(ACTIONS_RESULTS_URL))
+ .unwrap_or_default()
+ }
+ "v2" => {
+ // Only RESULTS_URL is used for v2
+ env::var(ACTIONS_RESULTS_URL).unwrap_or_default()
+ }
+ _ => panic!("Unsupported cache service version: {}", version),
+ }
+}
+
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct GhacQueryResponse {
(opendal) 02/03: Implement ghac v2 support
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch ghac-v2
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit de4ab91cfc049e4d7266560fc4e4896a848320b4
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Feb 17 18:19:05 2025 +0800
Implement ghac v2 support
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/Cargo.lock | 12 ++
core/Cargo.toml | 134 ++++++++++-----------
core/src/services/ghac/backend.rs | 55 +++------
core/src/services/ghac/core.rs | 243 +++++++++++++++++++++++---------------
core/src/services/ghac/error.rs | 32 ++++-
core/src/services/ghac/writer.rs | 47 ++++----
6 files changed, 294 insertions(+), 229 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index aa15df52a..8cbd440e3 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -3240,6 +3240,16 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "ghac"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "28847abc99c1a0f6897eac550967d435705e0ecd243c9c3b6b9e002557d73c42"
+dependencies = [
+ "prost 0.13.4",
+ "tonic 0.12.3",
+]
+
[[package]]
name = "ghost"
version = "0.1.18"
@@ -5143,6 +5153,7 @@ dependencies = [
"foundationdb",
"futures",
"getrandom 0.2.15",
+ "ghac",
"governor",
"hdfs-native",
"hdrs",
@@ -5192,6 +5203,7 @@ dependencies = [
"surrealdb",
"tikv-client",
"tokio",
+ "tonic 0.12.3",
"tracing",
"tracing-opentelemetry",
"tracing-subscriber",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 552029864..bada1d4c2 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -56,16 +56,16 @@ default = ["reqwest/rustls-tls", "executors-tokio", "services-memory"]
#
# You should never enable this feature unless you are developing opendal.
tests = [
- "dep:rand",
- "dep:sha2",
- "dep:dotenvy",
- "layers-blocking",
- "services-azblob",
- "services-fs",
- "services-http",
- "services-memory",
- "internal-tokio-rt",
- "services-s3",
+ "dep:rand",
+ "dep:sha2",
+ "dep:dotenvy",
+ "layers-blocking",
+ "services-azblob",
+ "services-fs",
+ "services-http",
+ "services-memory",
+ "internal-tokio-rt",
+ "services-s3",
]
# Enable path cache.
@@ -109,20 +109,20 @@ services-aliyun-drive = []
services-alluxio = []
services-atomicserver = ["dep:atomic_lib"]
services-azblob = [
- "dep:sha2",
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:sha2",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-azdls = [
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-azfile = [
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-b2 = []
services-cacache = ["dep:cacache"]
@@ -130,9 +130,9 @@ services-chainsafe = []
services-cloudflare-kv = []
services-compfs = ["dep:compio"]
services-cos = [
- "dep:reqsign",
- "reqsign?/services-tencent",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-tencent",
+ "reqsign?/reqwest_request",
]
services-d1 = []
services-dashmap = ["dep:dashmap"]
@@ -143,12 +143,12 @@ services-foundationdb = ["dep:foundationdb"]
services-fs = ["tokio/fs", "internal-tokio-rt"]
services-ftp = ["dep:suppaftp", "dep:bb8", "dep:async-tls"]
services-gcs = [
- "dep:reqsign",
- "reqsign?/services-google",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-google",
+ "reqsign?/reqwest_request",
]
services-gdrive = ["internal-path-cache"]
-services-ghac = []
+services-ghac = ["dep:ghac", "dep:tonic", "tonic/channel"]
services-github = []
services-gridfs = ["dep:mongodb", "dep:mongodb-internal-macros"]
services-hdfs = ["dep:hdrs"]
@@ -169,15 +169,15 @@ services-monoiofs = ["dep:monoio", "dep:flume"]
services-mysql = ["dep:sqlx", "sqlx?/mysql"]
services-nebula-graph = ["dep:rust-nebula", "dep:bb8", "dep:snowflaked"]
services-obs = [
- "dep:reqsign",
- "reqsign?/services-huaweicloud",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-huaweicloud",
+ "reqsign?/reqwest_request",
]
services-onedrive = []
services-oss = [
- "dep:reqsign",
- "reqsign?/services-aliyun",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-aliyun",
+ "reqsign?/reqwest_request",
]
services-pcloud = []
services-persy = ["dep:persy", "internal-tokio-rt"]
@@ -187,10 +187,10 @@ services-redis = ["dep:redis", "dep:bb8", "redis?/tokio-rustls-comp"]
services-redis-native-tls = ["services-redis", "redis?/tokio-native-tls-comp"]
services-rocksdb = ["dep:rocksdb", "internal-tokio-rt"]
services-s3 = [
- "dep:reqsign",
- "reqsign?/services-aws",
- "reqsign?/reqwest_request",
- "dep:crc32c",
+ "dep:reqsign",
+ "reqsign?/services-aws",
+ "reqsign?/reqwest_request",
+ "dep:crc32c",
]
services-seafile = []
services-sftp = ["dep:openssh", "dep:openssh-sftp-client", "dep:bb8"]
@@ -236,12 +236,12 @@ backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
bytes = "1.6"
chrono = { version = "0.4.28", default-features = false, features = [
- "clock",
- "std",
+ "clock",
+ "std",
] }
futures = { version = "0.3", default-features = false, features = [
- "std",
- "async-await",
+ "std",
+ "async-await",
] }
http = "1.1"
log = "0.4"
@@ -251,12 +251,14 @@ once_cell = "1"
percent-encoding = "2"
quick-xml = { version = "0.36", features = ["serialize", "overlapped-lists"] }
reqwest = { version = "0.12.2", features = [
- "stream",
+ "stream",
], default-features = false }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1.27", features = ["sync", "io-util"] }
uuid = { version = "1", features = ["serde", "v4"] }
+ghac = { version = "0.1.0", optional = true }
+tonic = { version = "0.12", optional = true, default-features = false }
# Test only dependencies
dotenvy = { version = "0.15", optional = true }
@@ -271,7 +273,7 @@ prost = { version = "0.13", optional = true }
sha1 = { version = "0.10.6", optional = true }
sha2 = { version = "0.10", optional = true }
sqlx = { version = "0.8.0", features = [
- "runtime-tokio-rustls",
+ "runtime-tokio-rustls",
], optional = true }
# For http based services.
@@ -284,8 +286,8 @@ ouroboros = { version = "0.18.4", optional = true }
atomic_lib = { version = "0.39.0", optional = true }
# for services-cacache
cacache = { version = "13.0", default-features = false, features = [
- "tokio-runtime",
- "mmap",
+ "tokio-runtime",
+ "mmap",
], optional = true }
# for services-dashmap
dashmap = { version = "6", optional = true }
@@ -293,8 +295,8 @@ dashmap = { version = "6", optional = true }
etcd-client = { version = "0.14", optional = true, features = ["tls"] }
# for services-foundationdb
foundationdb = { version = "0.9.0", features = [
- "embedded-fdb-include",
- "fdb-7_3",
+ "embedded-fdb-include",
+ "fdb-7_3",
], optional = true }
# for services-hdfs
hdrs = { version = "0.3.2", optional = true, features = ["async_file"] }
@@ -311,8 +313,8 @@ mongodb-internal-macros = { version = ">=3,<3.2.0", optional = true }
# for services-sftp
openssh = { version = "0.11.0", optional = true }
openssh-sftp-client = { version = "0.15.2", optional = true, features = [
- "openssh",
- "tracing",
+ "openssh",
+ "tracing",
] }
# for services-persy
persy = { version = "1.4.6", optional = true }
@@ -320,9 +322,9 @@ persy = { version = "1.4.6", optional = true }
redb = { version = "2", optional = true }
# for services-redis
redis = { version = "0.27", features = [
- "cluster-async",
- "tokio-comp",
- "connection-manager",
+ "cluster-async",
+ "tokio-comp",
+ "connection-manager",
], optional = true }
# for services-rocksdb
rocksdb = { version = "0.21.0", default-features = false, optional = true }
@@ -330,9 +332,9 @@ rocksdb = { version = "0.21.0", default-features = false, optional = true }
sled = { version = "0.34.7", optional = true }
# for services-ftp
suppaftp = { version = "6.0.3", default-features = false, features = [
- "async-secure",
- "rustls",
- "async-rustls",
+ "async-secure",
+ "rustls",
+ "async-rustls",
], optional = true }
# for services-tikv
tikv-client = { version = "0.3.0", optional = true, default-features = false }
@@ -342,10 +344,10 @@ hdfs-native = { version = "0.10", optional = true }
surrealdb = { version = "2", optional = true, features = ["protocol-http"] }
# for services-compfs
compio = { version = "0.12.0", optional = true, features = [
- "runtime",
- "bytes",
- "polling",
- "dispatcher",
+ "runtime",
+ "bytes",
+ "polling",
+ "dispatcher",
] }
# for services-s3
crc32c = { version = "0.6.6", optional = true }
@@ -355,10 +357,10 @@ snowflaked = { version = "1", optional = true, features = ["sync"] }
# for services-monoiofs
flume = { version = "0.11", optional = true }
monoio = { version = "0.2.4", optional = true, features = [
- "sync",
- "mkdirat",
- "unlinkat",
- "renameat",
+ "sync",
+ "mkdirat",
+ "unlinkat",
+ "renameat",
] }
# Layers
@@ -397,7 +399,7 @@ fastrace = { version = "0.7", features = ["enable"] }
fastrace-jaeger = "0.7"
libtest-mimic = "0.8"
opentelemetry = { version = "0.28", default-features = false, features = [
- "trace",
+ "trace",
] }
opentelemetry-otlp = { version = "0.28", features = ["grpc-tonic"] }
opentelemetry_sdk = { version = "0.28", features = ["rt-tokio"] }
@@ -408,6 +410,6 @@ size = "0.4"
tokio = { version = "1.27", features = ["fs", "macros", "rt-multi-thread"] }
tracing-opentelemetry = "0.29.0"
tracing-subscriber = { version = "0.3", features = [
- "env-filter",
- "tracing-log",
+ "env-filter",
+ "tracing-log",
] }
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index 64361c198..f3486f2dd 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -18,7 +18,6 @@
use std::env;
use std::sync::Arc;
-use bytes::Buf;
use http::header;
use http::Request;
use http::Response;
@@ -144,7 +143,7 @@ impl Builder for GhacBuilder {
let root = normalize_root(&self.config.root.unwrap_or_default());
debug!("backend use root {}", root);
- let client = if let Some(client) = self.http_client {
+ let http_client = if let Some(client) = self.http_client {
client
} else {
HttpClient::new().map_err(|err| {
@@ -168,7 +167,8 @@ impl Builder for GhacBuilder {
.clone()
.unwrap_or_else(|| "opendal".to_string()),
- client,
+ http_client,
+ grpc_client: None,
};
Ok(GhacBackend {
@@ -228,24 +228,13 @@ impl Access for GhacBackend {
///
/// In this way, we can support both self-hosted GHES and `github.com`.
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let req = self.core.ghac_query(path)?;
-
- let resp = self.core.client.send(req).await?;
-
- let location = if resp.status() == StatusCode::OK {
- let slc = resp.into_body();
- let query_resp: GhacQueryResponse =
- serde_json::from_reader(slc.reader()).map_err(new_json_deserialize_error)?;
- query_resp.archive_location
- } else {
- return Err(parse_error(resp));
- };
+ let location = self.core.ghac_get_download_url(path).await?;
let req = Request::get(location)
.header(header::RANGE, "bytes=0-0")
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = self.core.client.send(req).await?;
+ let resp = self.core.http_client.send(req).await?;
let status = resp.status();
match status {
@@ -266,21 +255,16 @@ impl Access for GhacBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let req = self.core.ghac_query(path)?;
+ let location = self.core.ghac_get_download_url(path).await?;
- let resp = self.core.client.send(req).await?;
+ let mut req = Request::get(location);
- let location = if resp.status() == StatusCode::OK {
- let slc = resp.into_body();
- let query_resp: GhacQueryResponse =
- serde_json::from_reader(slc.reader()).map_err(new_json_deserialize_error)?;
- query_resp.archive_location
- } else {
- return Err(parse_error(resp));
- };
+ if !args.range().is_full() {
+ req = req.header(header::RANGE, args.range().to_header());
+ }
+ let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
- let req = self.core.ghac_get_location(&location, args.range())?;
- let resp = self.core.client.fetch(req).await?;
+ let resp = self.core.http_client.fetch(req).await?;
let status = resp.status();
match status {
@@ -296,22 +280,11 @@ impl Access for GhacBackend {
}
async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let req = self.core.ghac_reserve(path)?;
-
- let resp = self.core.client.send(req).await?;
-
- let cache_id = if resp.status().is_success() {
- let slc = resp.into_body();
- let reserve_resp: GhacReserveResponse =
- serde_json::from_reader(slc.reader()).map_err(new_json_deserialize_error)?;
- reserve_resp.cache_id
- } else {
- return Err(parse_error(resp).map(|err| err.with_operation("Backend::ghac_reserve")));
- };
+ let url = self.core.ghac_get_upload_url(path).await?;
Ok((
RpWrite::default(),
- GhacWriter::new(self.core.clone(), cache_id),
+ GhacWriter::new(self.core.clone(), path.to_string(), url),
))
}
}
diff --git a/core/src/services/ghac/core.rs b/core/src/services/ghac/core.rs
index a9c6d5eb7..ea864f768 100644
--- a/core/src/services/ghac/core.rs
+++ b/core/src/services/ghac/core.rs
@@ -16,15 +16,19 @@
// under the License.
use crate::raw::{
- build_abs_path, new_json_serialize_error, new_request_build_error, percent_encode_path,
- BytesContentRange, BytesRange, HttpClient,
+ build_abs_path, new_json_deserialize_error, new_json_serialize_error, new_request_build_error,
+ percent_encode_path, HttpClient,
};
-use crate::Buffer;
-use bytes::Bytes;
-use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE};
-use http::{header, Request};
+use crate::services::ghac::error::{parse_error, parse_grpc_error};
+use crate::*;
+use ::ghac::v1 as ghac_grpc;
+use bytes::{Buf, Bytes};
+use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
+use http::{Request, StatusCode};
use serde::{Deserialize, Serialize};
use std::fmt::{Debug, Formatter};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
/// The base url for cache url.
pub const CACHE_URL_BASE: &str = "_apis/artifactcache";
@@ -50,7 +54,9 @@ pub struct GhacCore {
pub catch_token: String,
pub version: String,
- pub client: HttpClient,
+ pub http_client: HttpClient,
+
+ pub grpc_client: Option<ghac_grpc::cache_service_client::CacheServiceClient<Channel>>,
}
impl Debug for GhacCore {
@@ -64,106 +70,149 @@ impl Debug for GhacCore {
}
impl GhacCore {
- pub fn ghac_query(&self, path: &str) -> crate::Result<Request<Buffer>> {
+ pub async fn ghac_get_download_url(&self, path: &str) -> Result<String> {
let p = build_abs_path(&self.root, path);
- let url = format!(
- "{}{CACHE_URL_BASE}/cache?keys={}&version={}",
- self.cache_url,
- percent_encode_path(&p),
- self.version
- );
-
- let mut req = Request::get(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
-
- let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
-
- Ok(req)
- }
-
- pub fn ghac_get_location(
- &self,
- location: &str,
- range: BytesRange,
- ) -> crate::Result<Request<Buffer>> {
- let mut req = Request::get(location);
-
- if !range.is_full() {
- req = req.header(header::RANGE, range.to_header());
+ if let Some(grpc_client) = &self.grpc_client {
+ let req = ghac_grpc::GetCacheEntryDownloadUrlRequest {
+ key: p,
+ version: self.version.clone(),
+
+ metadata: None,
+ restore_keys: vec![],
+ };
+ let mut req = req.into_request();
+ req.metadata_mut().insert(
+ AUTHORIZATION.as_str(),
+ format!("Bearer {}", self.catch_token).parse().unwrap(),
+ );
+ let resp = grpc_client
+ .clone()
+ .get_cache_entry_download_url(req)
+ .await
+ .map_err(parse_grpc_error)?;
+ Ok(resp.into_inner().signed_download_url)
+ } else {
+ let url = format!(
+ "{}{CACHE_URL_BASE}/cache?keys={}&version={}",
+ self.cache_url,
+ percent_encode_path(&p),
+ self.version
+ );
+
+ let mut req = Request::get(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+
+ let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
+ let resp = self.http_client.send(req).await?;
+ let location = if resp.status() == StatusCode::OK {
+ let slc = resp.into_body();
+ let query_resp: GhacQueryResponse =
+ serde_json::from_reader(slc.reader()).map_err(new_json_deserialize_error)?;
+ query_resp.archive_location
+ } else {
+ return Err(parse_error(resp));
+ };
+ Ok(location)
}
-
- req.body(Buffer::new()).map_err(new_request_build_error)
}
- pub fn ghac_reserve(&self, path: &str) -> crate::Result<Request<Buffer>> {
+ pub async fn ghac_get_upload_url(&self, path: &str) -> Result<String> {
let p = build_abs_path(&self.root, path);
- let url = format!("{}{CACHE_URL_BASE}/caches", self.cache_url);
-
- let bs = serde_json::to_vec(&GhacReserveRequest {
- key: p,
- version: self.version.to_string(),
- })
- .map_err(new_json_serialize_error)?;
-
- let mut req = Request::post(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_LENGTH, bs.len());
- req = req.header(CONTENT_TYPE, "application/json");
-
- let req = req
- .body(Buffer::from(Bytes::from(bs)))
- .map_err(new_request_build_error)?;
-
- Ok(req)
- }
-
- pub fn ghac_upload(
- &self,
- cache_id: i64,
- offset: u64,
- size: u64,
- body: Buffer,
- ) -> crate::Result<Request<Buffer>> {
- let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
-
- let mut req = Request::patch(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_LENGTH, size);
- req = req.header(CONTENT_TYPE, "application/octet-stream");
- req = req.header(
- CONTENT_RANGE,
- BytesContentRange::default()
- .with_range(offset, offset + size - 1)
- .to_header(),
- );
-
- let req = req.body(body).map_err(new_request_build_error)?;
-
- Ok(req)
+ if let Some(grpc_client) = &self.grpc_client {
+ let req = ghac_grpc::CreateCacheEntryRequest {
+ key: p,
+ version: self.version.clone(),
+
+ metadata: None,
+ };
+ let mut req = req.into_request();
+ req.metadata_mut().insert(
+ AUTHORIZATION.as_str(),
+ format!("Bearer {}", self.catch_token).parse().unwrap(),
+ );
+ let resp = grpc_client
+ .clone()
+ .create_cache_entry(req)
+ .await
+ .map_err(parse_grpc_error)?;
+ Ok(resp.into_inner().signed_upload_url)
+ } else {
+ let url = format!("{}{CACHE_URL_BASE}/caches", self.cache_url);
+
+ let bs = serde_json::to_vec(&GhacReserveRequest {
+ key: p,
+ version: self.version.to_string(),
+ })
+ .map_err(new_json_serialize_error)?;
+
+ let mut req = Request::post(&url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+ req = req.header(CONTENT_LENGTH, bs.len());
+ req = req.header(CONTENT_TYPE, "application/json");
+
+ let req = req
+ .body(Buffer::from(Bytes::from(bs)))
+ .map_err(new_request_build_error)?;
+ let resp = self.http_client.send(req).await?;
+ let cache_id = if resp.status().is_success() {
+ let slc = resp.into_body();
+ let reserve_resp: GhacReserveResponse =
+ serde_json::from_reader(slc.reader()).map_err(new_json_deserialize_error)?;
+ reserve_resp.cache_id
+ } else {
+ return Err(
+ parse_error(resp).map(|err| err.with_operation("Backend::ghac_reserve"))
+ );
+ };
+
+ let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
+ Ok(url)
+ }
}
- pub fn ghac_commit(&self, cache_id: i64, size: u64) -> crate::Result<Request<Buffer>> {
- let url = format!("{}{CACHE_URL_BASE}/caches/{cache_id}", self.cache_url);
-
- let bs =
- serde_json::to_vec(&GhacCommitRequest { size }).map_err(new_json_serialize_error)?;
-
- let mut req = Request::post(&url);
- req = req.header(AUTHORIZATION, format!("Bearer {}", self.catch_token));
- req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
- req = req.header(CONTENT_TYPE, "application/json");
- req = req.header(CONTENT_LENGTH, bs.len());
-
- let req = req
- .body(Buffer::from(Bytes::from(bs)))
- .map_err(new_request_build_error)?;
+ pub async fn ghac_finalize_upload(&self, path: &str, url: &str, size: u64) -> Result<()> {
+ let p = build_abs_path(&self.root, path);
- Ok(req)
+ if let Some(grpc_client) = &self.grpc_client {
+ let req = ghac_grpc::FinalizeCacheEntryUploadRequest {
+ key: p,
+ version: self.version.clone(),
+ size_bytes: size as i64,
+
+ metadata: None,
+ };
+ let mut req = req.into_request();
+ req.metadata_mut().insert(
+ AUTHORIZATION.as_str(),
+ format!("Bearer {}", self.catch_token).parse().unwrap(),
+ );
+ let _ = grpc_client
+ .clone()
+ .finalize_cache_entry_upload(req)
+ .await
+ .map_err(parse_grpc_error)?;
+ Ok(())
+ } else {
+ let bs = serde_json::to_vec(&GhacCommitRequest { size })
+ .map_err(new_json_serialize_error)?;
+
+ let req = Request::post(url)
+ .header(AUTHORIZATION, format!("Bearer {}", self.catch_token))
+ .header(ACCEPT, CACHE_HEADER_ACCEPT)
+ .header(CONTENT_LENGTH, bs.len())
+ .body(Buffer::from(bs))
+ .map_err(new_request_build_error)?;
+ let resp = self.http_client.send(req).await?;
+ if resp.status().is_success() {
+ Ok(())
+ } else {
+ Err(parse_error(resp))
+ }
+ }
}
}
diff --git a/core/src/services/ghac/error.rs b/core/src/services/ghac/error.rs
index 7efafe1f6..c4c1b7d3f 100644
--- a/core/src/services/ghac/error.rs
+++ b/core/src/services/ghac/error.rs
@@ -15,11 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use http::Response;
-use http::StatusCode;
-
use crate::raw::*;
use crate::*;
+use http::Response;
+use http::StatusCode;
+use tonic::{Code, Status};
/// Parse error response into Error.
pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
@@ -50,3 +50,29 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
err
}
+
+pub fn parse_grpc_error(status: Status) -> Error {
+ let (kind, retryable) = match status.code() {
+ Code::Ok => unreachable!("parse_grpc_error should not be called with Code::Ok"),
+ Code::NotFound => (ErrorKind::NotFound, false),
+ Code::AlreadyExists => (ErrorKind::AlreadyExists, false),
+ Code::PermissionDenied => (ErrorKind::PermissionDenied, false),
+ Code::FailedPrecondition => (ErrorKind::ConditionNotMatch, false),
+ Code::Cancelled
+ | Code::DeadlineExceeded
+ | Code::ResourceExhausted
+ | Code::Aborted
+ | Code::Internal
+ | Code::Unavailable => (ErrorKind::Unexpected, true),
+ _ => (ErrorKind::Unexpected, false),
+ };
+
+ let mut err = Error::new(kind, status.message())
+ .with_context("metadata", format!("{:?}", status.metadata()));
+
+ if retryable {
+ err = err.set_temporary();
+ }
+
+ err
+}
diff --git a/core/src/services/ghac/writer.rs b/core/src/services/ghac/writer.rs
index 90b39c564..d60e345cc 100644
--- a/core/src/services/ghac/writer.rs
+++ b/core/src/services/ghac/writer.rs
@@ -19,20 +19,24 @@ use super::core::*;
use super::error::parse_error;
use crate::raw::*;
use crate::*;
+use http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE};
+use http::Request;
use std::sync::Arc;
pub struct GhacWriter {
core: Arc<GhacCore>,
- cache_id: i64,
+ path: String,
+ url: String,
size: u64,
}
impl GhacWriter {
- pub fn new(core: Arc<GhacCore>, cache_id: i64) -> Self {
+ pub fn new(core: Arc<GhacCore>, path: String, url: String) -> Self {
GhacWriter {
core,
- cache_id,
+ path,
+ url,
size: 0,
}
}
@@ -40,23 +44,27 @@ impl GhacWriter {
impl oio::Write for GhacWriter {
async fn write(&mut self, bs: Buffer) -> Result<()> {
- let size = bs.len();
+ let size = bs.len() as u64;
let offset = self.size;
- let req = self.core.ghac_upload(
- self.cache_id,
- offset,
- size as u64,
- Buffer::from(bs.to_bytes()),
- )?;
-
- let resp = self.core.client.send(req).await?;
+ let mut req = Request::patch(&self.url);
+ req = req.header(AUTHORIZATION, format!("Bearer {}", self.core.catch_token));
+ req = req.header(ACCEPT, CACHE_HEADER_ACCEPT);
+ req = req.header(CONTENT_LENGTH, size);
+ req = req.header(CONTENT_TYPE, "application/octet-stream");
+ req = req.header(
+ CONTENT_RANGE,
+ BytesContentRange::default()
+ .with_range(offset, offset + size - 1)
+ .to_header(),
+ );
+ let req = req.body(bs).map_err(new_request_build_error)?;
+ let resp = self.core.http_client.send(req).await?;
if !resp.status().is_success() {
return Err(parse_error(resp).map(|err| err.with_operation("Backend::ghac_upload")));
}
-
- self.size += size as u64;
+ self.size += size;
Ok(())
}
@@ -65,13 +73,8 @@ impl oio::Write for GhacWriter {
}
async fn close(&mut self) -> Result<()> {
- let req = self.core.ghac_commit(self.cache_id, self.size)?;
- let resp = self.core.client.send(req).await?;
-
- if resp.status().is_success() {
- Ok(())
- } else {
- Err(parse_error(resp).map(|err| err.with_operation("Backend::ghac_commit")))
- }
+ self.core
+ .ghac_finalize_upload(&self.path, &self.url, self.size)
+ .await
}
}