Skip to content

Commit 8b35a9f

Browse files
author
Devdutt Shenoi
authored
Merge branch 'main' into query-param
2 parents 1138006 + 43a793d commit 8b35a9f

14 files changed

+487
-198
lines changed

.github/workflows/build-push-edge-kafka.yaml

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,11 +37,24 @@ jobs:
3737
with:
3838
images: parseable/parseable
3939

40-
- name: Build and push
40+
- name: Build and push x86_64
4141
uses: docker/build-push-action@ad44023a93711e3deb337508980b4b5e9bcdc5dc
4242
with:
4343
context: .
4444
file: ./Dockerfile.kafka
4545
push: true
4646
tags: parseable/parseable:edge-kafka
47-
platforms: linux/amd64,linux/arm64
47+
platforms: linux/amd64
48+
build-args: |
49+
LIB_DIR=x86_64-linux-gnu
50+
51+
- name: Build and push aarch64
52+
uses: docker/build-push-action@ad44023a93711e3deb337508980b4b5e9bcdc5dc
53+
with:
54+
context: .
55+
file: ./Dockerfile.kafka
56+
push: true
57+
tags: parseable/parseable:edge-kafka
58+
platforms: =linux/arm64
59+
build-args: |
60+
LIB_DIR=aarch64-linux-gnu

.github/workflows/integration-test.yaml

Lines changed: 1 addition & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -32,27 +32,4 @@ jobs:
3232
- name: Stop compose
3333
if: always()
3434
run: docker compose -f docker-compose-distributed-test.yaml down -v
35-
36-
docker-compose-test-with-kafka:
37-
name: Quest Smoke and Load Tests for Standalone deployments with Kafka
38-
runs-on: ubuntu-latest
39-
steps:
40-
- name: Checkout
41-
uses: actions/checkout@v4
42-
- name: Start compose
43-
run: docker compose -f docker-compose-test-with-kafka.yaml up --build --exit-code-from quest
44-
- name: Stop compose
45-
if: always()
46-
run: docker compose -f docker-compose-test-with-kafka.yaml down -v
47-
48-
docker-compose-distributed-test-with-kafka:
49-
name: Quest Smoke and Load Tests for Distributed deployments with Kafka
50-
runs-on: ubuntu-latest
51-
steps:
52-
- name: Checkout
53-
uses: actions/checkout@v4
54-
- name: Start compose
55-
run: docker compose -f docker-compose-distributed-test-with-kafka.yaml up --build --exit-code-from quest
56-
- name: Stop compose
57-
if: always()
58-
run: docker compose -f docker-compose-distributed-test-with-kafka.yaml down -v
35+

Dockerfile.kafka

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -49,16 +49,17 @@ RUN cargo build --release --features kafka
4949
# final stage
5050
FROM gcr.io/distroless/cc-debian12:latest
5151

52+
# Copy only the libraries that binary needs since kafka is statically linked
53+
ARG LIB_DIR
54+
COPY --from=builder /usr/lib/${LIB_DIR}/libsasl2.so.2 /usr/lib/${LIB_DIR}/
55+
COPY --from=builder /usr/lib/${LIB_DIR}/libssl.so.3 /usr/lib/${LIB_DIR}/
56+
COPY --from=builder /usr/lib/${LIB_DIR}/libcrypto.so.3 /usr/lib/${LIB_DIR}/
57+
5258
WORKDIR /parseable
5359

5460
# Copy the Parseable binary from builder
5561
COPY --from=builder /parseable/target/release/parseable /usr/bin/parseable
5662

57-
# Copy only the libraries that binary needs since kafka is statically linked
58-
COPY --from=builder /usr/lib/x86_64-linux-gnu/libsasl2.so.2 /usr/lib/x86_64-linux-gnu/
59-
COPY --from=builder /usr/lib/x86_64-linux-gnu/libssl.so.3 /usr/lib/x86_64-linux-gnu/
60-
COPY --from=builder /usr/lib/x86_64-linux-gnu/libcrypto.so.3 /usr/lib/x86_64-linux-gnu/
61-
6263
# Copy CA certificates
6364
COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
6465

docker-compose-distributed-test-with-kafka.yaml

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,8 @@ services:
9494
build:
9595
context: .
9696
dockerfile: Dockerfile.kafka
97+
args:
98+
- LIB_DIR=x86_64-linux-gnu
9799
platform: linux/amd64
98100
command: [ "parseable", "s3-store", ]
99101
expose:
@@ -140,6 +142,8 @@ services:
140142
build:
141143
context: .
142144
dockerfile: Dockerfile.kafka
145+
args:
146+
- LIB_DIR=x86_64-linux-gnu
143147
platform: linux/amd64
144148
command: [ "parseable", "s3-store", ]
145149
expose:

docker-compose-test-with-kafka.yaml

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@ services:
2828
build:
2929
context: .
3030
dockerfile: Dockerfile.kafka
31+
args:
32+
- LIB_DIR=x86_64-linux-gnu
3133
platform: linux/amd64
3234
command: [ "parseable", "s3-store", ]
3335
ports:

src/alerts/mod.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -330,7 +330,7 @@ pub struct RollingWindow {
330330
// should always be "now"
331331
pub eval_end: String,
332332
// x minutes (5m)
333-
pub eval_frequency: u32,
333+
pub eval_frequency: u64,
334334
}
335335

336336
#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
@@ -641,7 +641,7 @@ impl AlertConfig {
641641
columns
642642
}
643643

644-
pub fn get_eval_frequency(&self) -> u32 {
644+
pub fn get_eval_frequency(&self) -> u64 {
645645
match &self.eval_type {
646646
EvalConfig::RollingWindow(rolling_window) => rolling_window.eval_frequency,
647647
}

src/handlers/http/logstream.rs

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -103,8 +103,6 @@ pub async fn list(req: HttpRequest) -> Result<impl Responder, StreamError> {
103103
.unwrap()
104104
.into_iter()
105105
.filter(|logstream| {
106-
warn!("logstream-\n{logstream:?}");
107-
108106
Users.authorize(key.clone(), Action::ListStream, Some(logstream), None)
109107
== crate::rbac::Response::Authorized
110108
})

src/handlers/http/modal/ingest_server.rs

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,11 @@ impl ParseableServer for IngestServer {
203203
sync::run_local_sync().await;
204204
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
205205
sync::object_store_sync().await;
206+
let (
207+
mut remote_conversion_handler,
208+
mut remote_conversion_outbox,
209+
mut remote_conversion_inbox,
210+
) = sync::arrow_conversion().await;
206211

207212
tokio::spawn(airplane::server());
208213

@@ -219,12 +224,16 @@ impl ParseableServer for IngestServer {
219224
// actix server finished .. stop other threads and stop the server
220225
remote_sync_inbox.send(()).unwrap_or(());
221226
localsync_inbox.send(()).unwrap_or(());
227+
remote_conversion_inbox.send(()).unwrap_or(());
222228
if let Err(e) = localsync_handler.await {
223229
error!("Error joining remote_sync_handler: {:?}", e);
224230
}
225231
if let Err(e) = remote_sync_handler.await {
226232
error!("Error joining remote_sync_handler: {:?}", e);
227233
}
234+
if let Err(e) = remote_conversion_handler.await {
235+
error!("Error joining remote_conversion_handler: {:?}", e);
236+
}
228237
return e
229238
},
230239
_ = &mut localsync_outbox => {
@@ -238,6 +247,13 @@ impl ParseableServer for IngestServer {
238247
error!("Error joining remote_sync_handler: {:?}", e);
239248
}
240249
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
250+
},
251+
_ = &mut remote_conversion_outbox => {
252+
// remote_conversion failed, this is recoverable by just starting remote_conversion thread again
253+
if let Err(e) = remote_conversion_handler.await {
254+
error!("Error joining remote_conversion_handler: {:?}", e);
255+
}
256+
(remote_conversion_handler, remote_conversion_outbox, remote_conversion_inbox) = sync::arrow_conversion().await;
241257
}
242258

243259
}

src/handlers/http/modal/mod.rs

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,19 @@ pub trait ParseableServer {
139139

140140
// Perform S3 sync and wait for completion
141141
info!("Starting data sync to S3...");
142-
if let Err(e) = CONFIG.storage().get_object_store().sync(true).await {
142+
143+
if let Err(e) = CONFIG.storage().get_object_store().conversion(true).await {
144+
warn!("Failed to convert arrow files to parquet. {:?}", e);
145+
} else {
146+
info!("Successfully converted arrow files to parquet.");
147+
}
148+
149+
if let Err(e) = CONFIG
150+
.storage()
151+
.get_object_store()
152+
.upload_files_from_staging()
153+
.await
154+
{
143155
warn!("Failed to sync local data with object store. {:?}", e);
144156
} else {
145157
info!("Successfully synced all data to S3.");

src/handlers/http/modal/server.rs

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,11 @@ impl ParseableServer for Server {
134134
sync::run_local_sync().await;
135135
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
136136
sync::object_store_sync().await;
137-
137+
let (
138+
mut remote_conversion_handler,
139+
mut remote_conversion_outbox,
140+
mut remote_conversion_inbox,
141+
) = sync::arrow_conversion().await;
138142
if CONFIG.options.send_analytics {
139143
analytics::init_analytics_scheduler()?;
140144
}
@@ -152,12 +156,16 @@ impl ParseableServer for Server {
152156
// actix server finished .. stop other threads and stop the server
153157
remote_sync_inbox.send(()).unwrap_or(());
154158
localsync_inbox.send(()).unwrap_or(());
159+
remote_conversion_inbox.send(()).unwrap_or(());
155160
if let Err(e) = localsync_handler.await {
156161
error!("Error joining remote_sync_handler: {:?}", e);
157162
}
158163
if let Err(e) = remote_sync_handler.await {
159164
error!("Error joining remote_sync_handler: {:?}", e);
160165
}
166+
if let Err(e) = remote_conversion_handler.await {
167+
error!("Error joining remote_conversion_handler: {:?}", e);
168+
}
161169
return e
162170
},
163171
_ = &mut localsync_outbox => {
@@ -171,6 +179,13 @@ impl ParseableServer for Server {
171179
error!("Error joining remote_sync_handler: {:?}", e);
172180
}
173181
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
182+
},
183+
_ = &mut remote_conversion_outbox => {
184+
// remote_conversion failed, this is recoverable by just starting remote_conversion thread again
185+
if let Err(e) = remote_conversion_handler.await {
186+
error!("Error joining remote_conversion_handler: {:?}", e);
187+
}
188+
(remote_conversion_handler, remote_conversion_outbox, remote_conversion_inbox) = sync::arrow_conversion().await;
174189
}
175190

176191
};

0 commit comments

Comments
 (0)