diff --git a/Cargo.toml b/Cargo.toml index 302de69..912567a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,5 +13,6 @@ repository = "https://github.com/aesteve/buska" [workspace] members = [ "./core", - "./cli" + "./cli", + "./web" ] diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 653b553..8845b38 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -17,5 +17,5 @@ indicatif = "0.17.0-beta.1" chrono = "0.4.19" rdkafka = { version = "0.27" , features = ["ssl", "cmake-build" ] } serde_json = "1.0" -tokio = { version = "1.3.0", features = ["macros", "rt", "time", "sync", "rt-multi-thread"] } +tokio = { version = "1.14.0", features = ["macros", "rt", "time", "sync", "rt-multi-thread"] } futures = "0.3" \ No newline at end of file diff --git a/core/Cargo.toml b/core/Cargo.toml index 35b273b..67f427b 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -10,7 +10,7 @@ keywords = ["kafka", "apache kafka", "search", "lib"] rdkafka = { version = "0.27" , features = ["ssl", "cmake-build" ] } serde = { version = "1.0.124", features = ["derive"] } chrono = { version = "0.4.19", features = ["serde"] } -tokio = { version = "1.3.0", features = ["macros", "rt", "time", "sync"] } +tokio = { version = "1.14.0", features = ["macros", "rt", "time", "sync"] } log = "0.4" env_logger = "0.9" jsonpath-rust = "0.1.3" diff --git a/core/src/lib.rs b/core/src/lib.rs index 5ce9f43..33a3055 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -555,7 +555,7 @@ mod tests { //clean("test").await; //prepare("test", 60).await; let before = Utc::now(); - // let range = 1..1_000_000; + let range = 1..1_000_000; // let range = 1_000_001..2_000_000; // let range = 2_000_001..3_000_000; // let range = 3_000_001..4_000_000; @@ -565,7 +565,7 @@ mod tests { // let range = 6_000_001..7_000_000; // let range = 7_000_001..8_000_000; // let range = 8_000_001..9_000_000; - let range = 9_000_001..10_000_000; + // let range = 9_000_001..10_000_000; let recs = range.into_iter().map(|i| TestRecord { key: format!("key-{}",i), nested: NestedTestRecord { int: i, ints: vec![i-1, i, i+1], diff --git a/core/src/search/notifications.rs b/core/src/search/notifications.rs index 17d9cba..afc8f3b 100644 --- a/core/src/search/notifications.rs +++ b/core/src/search/notifications.rs @@ -2,6 +2,7 @@ use std::collections::BTreeMap; use std::fmt::{Display, Formatter}; use chrono::{DateTime, Duration, Utc}; use rdkafka::message::OwnedMessage; +use serde::Serialize; #[derive(Debug, Clone)] pub enum SearchNotification { @@ -20,7 +21,7 @@ pub enum PreparationStep { SeekPartitions, } -#[derive(Debug, Clone, PartialEq)] +#[derive(Serialize, Debug, Clone, PartialEq)] pub struct Progress { pub done: i64, pub total: i64, diff --git a/web/Cargo.toml b/web/Cargo.toml new file mode 100644 index 0000000..68bc412 --- /dev/null +++ b/web/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "buska-web" +version = "0.0.1" +edition = "2021" +authors = ["aesteve "] +description = "A web interface for Buska: a tool to look for data in Kafka topics" +keywords = ["kafka", "apache kafka", "search", "web"] + +[dependencies] +buska-core = { path = "../core" } +config = "0.11.0" +env_logger = "0.9.0" +axum = { version = "0.3.4" } +hyper = { version = "0.14.15", features = ["full"] } +tokio = { version = "1.14.0", features = ["full"] } +tokio-stream = "0.1" +tower = { version = "0.4.11", features = ["full"] } +tower-http = { version = "0.1.2", features = ["fs", "trace"] } + +# common dependencies for whole project +chrono = "0.4.19" +rdkafka = { version = "0.27" , features = ["ssl", "cmake-build" ] } +serde_json = "1.0" +serde = { version = "1.0.124", features = ["derive"] } +futures = "0.3" \ No newline at end of file diff --git a/web/assets/css/main.css b/web/assets/css/main.css new file mode 100644 index 0000000..e69de29 diff --git a/web/assets/index.html b/web/assets/index.html new file mode 100644 index 0000000..c6e14d9 --- /dev/null +++ b/web/assets/index.html @@ -0,0 +1,75 @@ + + + + + + + + + + BusKa Web + + + +
+ + + + \ No newline at end of file diff --git a/web/front/dev.sh b/web/front/dev.sh new file mode 100644 index 0000000..7074a38 --- /dev/null +++ b/web/front/dev.sh @@ -0,0 +1 @@ +elm-live src/Home.elm -- --output='../assets/elm.js' diff --git a/web/front/elm.json b/web/front/elm.json new file mode 100644 index 0000000..7b3f7a6 --- /dev/null +++ b/web/front/elm.json @@ -0,0 +1,27 @@ +{ + "type": "application", + "source-directories": [ + "src" + ], + "elm-version": "0.19.1", + "dependencies": { + "direct": { + "cuducos/elm-format-number": "9.0.1", + "elm/browser": "1.0.2", + "elm/core": "1.0.5", + "elm/html": "1.0.0", + "elm/json": "1.1.3" + }, + "indirect": { + "elm/regex": "1.0.0", + "elm/time": "1.0.0", + "elm/url": "1.0.0", + "elm/virtual-dom": "1.0.2", + "myrho/elm-round": "1.0.4" + } + }, + "test-dependencies": { + "direct": {}, + "indirect": {} + } +} diff --git a/web/front/src/Commands.elm b/web/front/src/Commands.elm new file mode 100644 index 0000000..c256cd3 --- /dev/null +++ b/web/front/src/Commands.elm @@ -0,0 +1,37 @@ +module Commands exposing (..) + +import Json.Decode exposing (Error) +import Model exposing (Progress, SearchProgress, SearchStep, decodeSearchProgress, decodeSearchStep) + + +type Msg + = StartSearch + | MatchFound String + | SearchStepChanged SearchStep + | SearchProgressed SearchProgress + | DecodingError Error + + +receivedSearchProgress : Json.Decode.Value -> Msg +receivedSearchProgress json = + case Json.Decode.decodeValue decodeSearchProgress json of + Ok decoded -> + SearchProgressed decoded + + Err message -> + DecodingError message + + +updateMatches : String -> Msg +updateMatches m = + MatchFound m + + +receivedSearchStep : Json.Decode.Value -> Msg +receivedSearchStep json = + case Json.Decode.decodeValue decodeSearchStep json of + Ok decoded -> + SearchStepChanged decoded + + Err message -> + DecodingError message diff --git a/web/front/src/Home.elm b/web/front/src/Home.elm new file mode 100644 index 0000000..71cceae --- /dev/null +++ b/web/front/src/Home.elm @@ -0,0 +1,68 @@ +module Home exposing (main) + +import Browser +import Commands exposing (Msg(..), receivedSearchProgress, receivedSearchStep, updateMatches) +import Html exposing (..) +import Html.Attributes exposing (..) +import Model exposing (..) +import Ports.JSSearch exposing (matchFound, searchProgressed, searchStepChanged, startSearch) +import Views.SearchForm exposing (searchForm) +import Views.SearchProgress exposing (searchProgress) +import Views.SearchResults exposing (searchResults) + + +view : State -> Html Msg +view state = + div [ class "main" ] + [ h1 [] [ text "Welcome to BusKa" ] + , searchForm state + , searchProgress state + , searchResults state + ] + + +init : () -> ( State, Cmd Msg ) +init _ = + ( { definition = Nothing, results = resetSearchResults }, Cmd.none ) + + +update : Msg -> State -> ( State, Cmd Msg ) +update msg state = + case msg of + StartSearch -> + ( state, startSearch () ) + + SearchProgressed searchProgress -> + ( { state | results = updateProgress state.results searchProgress }, Cmd.none ) + + SearchStepChanged newSearchStep -> + ( { state | results = updateStep state.results newSearchStep }, Cmd.none ) + + DecodingError error -> + let + _ = + Debug.log "Error in decoding JSON " error + in + ( state, Cmd.none ) + + MatchFound string -> + ( { state | results = appendMatch state.results string }, Cmd.none ) + + +subscriptions : State -> Sub Msg +subscriptions state = + Sub.batch + [ searchProgressed receivedSearchProgress + , matchFound updateMatches + , searchStepChanged receivedSearchStep + ] + + +main : Program () State Msg +main = + Browser.element + { init = init + , view = view + , update = update + , subscriptions = subscriptions + } diff --git a/web/front/src/Model.elm b/web/front/src/Model.elm new file mode 100644 index 0000000..d5cd624 --- /dev/null +++ b/web/front/src/Model.elm @@ -0,0 +1,101 @@ +module Model exposing (..) + +import Dict exposing (Dict) +import Json.Decode as D + + +resetSearchResults : SearchResults +resetSearchResults = + { matches = [] + , progress = NotStarted + , steps = [] + } + + +updateProgress : SearchResults -> SearchProgress -> SearchResults +updateProgress results progress = + { results | progress = InProgress progress } + + +updateStep : SearchResults -> SearchStep -> SearchResults +updateStep results step = + if step.step == 6 then + { results | steps = results.steps ++ [ step ], progress = Finished } + + else + { results | steps = results.steps ++ [ step ] } + + +appendMatch : SearchResults -> String -> SearchResults +appendMatch results newMatch = + { results | matches = results.matches ++ [ newMatch ] } + + +type alias SearchResults = + { matches : List String + , progress : SearchState + , steps : List SearchStep + } + + +type SearchState + = NotStarted + | InProgress SearchProgress + | Finished + + +type alias SearchDefinition = + { todo : String } + + +type alias SearchStep = + { step : Int + , description : String + } + + +type alias State = + { definition : Maybe SearchDefinition + , results : SearchResults + } + + +type alias Progress = + { done : Int + , total : Int + , rate : Float + } + + +type alias SearchProgress = + { elapsed : Int + , eta : String + , matches : Int + , overallProgress : Progress + , perPartitionProgress : Dict String Progress + } + + +decodeProgress : D.Decoder Progress +decodeProgress = + D.map3 Progress + (D.field "done" D.int) + (D.field "total" D.int) + (D.field "rate" D.float) + + +decodeSearchProgress : D.Decoder SearchProgress +decodeSearchProgress = + D.map5 SearchProgress + (D.field "elapsed" D.int) + (D.field "eta" D.string) + (D.field "matches" D.int) + (D.field "overall_progress" decodeProgress) + (D.field "per_partition_progress" (D.dict decodeProgress)) + + +decodeSearchStep : D.Decoder SearchStep +decodeSearchStep = + D.map2 SearchStep + (D.field "step" D.int) + (D.field "description" D.string) diff --git a/web/front/src/Ports/JSSearch.elm b/web/front/src/Ports/JSSearch.elm new file mode 100644 index 0000000..cdc99c1 --- /dev/null +++ b/web/front/src/Ports/JSSearch.elm @@ -0,0 +1,19 @@ +port module Ports.JSSearch exposing (..) + +import Json.Decode + + +port startSearch : () -> Cmd msg + + +port searchProgressed : (Json.Decode.Value -> msg) -> Sub msg + + +port matchFound : (String -> msg) -> Sub msg + + +port searchStepChanged : (Json.Decode.Value -> msg) -> Sub msg + + + +--port searchError: diff --git a/web/front/src/Views/SearchForm.elm b/web/front/src/Views/SearchForm.elm new file mode 100644 index 0000000..1876e0c --- /dev/null +++ b/web/front/src/Views/SearchForm.elm @@ -0,0 +1,37 @@ +module Views.SearchForm exposing (..) + +import Commands exposing (Msg(..)) +import Html exposing (..) +import Html.Attributes exposing (..) +import Html.Events exposing (onClick) +import Model exposing (State) + + +searchForm : State -> Html Msg +searchForm state = + div [] + [ div [ class "card" ] + [ div [ class "card-divider" ] [ text "Search options" ] + , div [ class "card-section" ] + [ div [ class "grid-container full" ] + [ div [ class "grid-x grid-padding-x" ] + [ div [ class "medium-9 cell" ] + [ label [ for "host" ] + [ text "Host" + , input [ type_ "text", id "host", placeholder "localhost" ] [] + ] + ] + , div [ class "medium-3 cell" ] + [ label [ for "port" ] + [ text "Port" + , input [ type_ "number", id "port", placeholder "9092" ] [] + ] + ] + ] + ] + ] + , div [ class "card-section" ] + [ button [ class "button", onClick StartSearch ] [ text "Search" ] + ] + ] + ] diff --git a/web/front/src/Views/SearchProgress.elm b/web/front/src/Views/SearchProgress.elm new file mode 100644 index 0000000..381dac6 --- /dev/null +++ b/web/front/src/Views/SearchProgress.elm @@ -0,0 +1,72 @@ +module Views.SearchProgress exposing (..) + +import Commands exposing (Msg) +import Dict +import FormatNumber exposing (format) +import FormatNumber.Locales exposing (Decimals(..), System(..), usLocale) +import Html exposing (..) +import Html.Attributes exposing (..) +import Model exposing (Progress, SearchResults, SearchState(..), SearchStep, State) + + +percentFormat = + { usLocale | decimals = Exact 2 } + + +searchProgress : State -> Html Msg +searchProgress state = + div + [] + [ searchSteps state.results + , searchProgressBar state.results + ] + + +progressBar : Progress -> Html Msg +progressBar p = + progress + [ class "full-progress-blue" + , Html.Attributes.min "0" + , p.done |> String.fromInt |> value + , p.total |> String.fromInt |> Html.Attributes.max + ] + [ text (format percentFormat (p.rate * 100) ++ "%") ] + + +partitionProgressBar : String -> Progress -> List (Html Msg) -> List (Html Msg) +partitionProgressBar part p list = + list ++ [ li [] [ span [] [ text (part ++ ":") ], progressBar p ] ] + + +searchProgressBar : SearchResults -> Html Msg +searchProgressBar state = + div [] + (case state.progress of + InProgress currentProgress -> + [ progressBar currentProgress.overallProgress + , ul [ class "no-style-list" ] (Dict.foldl partitionProgressBar [] currentProgress.perPartitionProgress) + ] + + Finished -> + [ progress + [ class "full-progress-blue" + , Html.Attributes.min "0" + , value "100" + , Html.Attributes.max "100" + ] + [ text "100%" ] + ] + + _ -> + [] + ) + + +searchSteps : SearchResults -> Html Msg +searchSteps results = + ul [ class "no-style-list" ] (List.map searchStep results.steps) + + +searchStep : SearchStep -> Html Msg +searchStep step = + li [] [ text (String.fromInt step.step ++ ". " ++ step.description) ] diff --git a/web/front/src/Views/SearchResults.elm b/web/front/src/Views/SearchResults.elm new file mode 100644 index 0000000..49d2715 --- /dev/null +++ b/web/front/src/Views/SearchResults.elm @@ -0,0 +1,16 @@ +module Views.SearchResults exposing (..) + +import Commands exposing (Msg) +import Html exposing (..) +import Html.Attributes exposing (..) +import Model exposing (State) + + +searchResults : State -> Html Msg +searchResults state = + div [ id "matched" ] (List.map display_matching state.results.matches) + + +display_matching : String -> Html msg +display_matching matched = + textarea [ disabled True ] [ text matched ] diff --git a/web/src/main.rs b/web/src/main.rs new file mode 100644 index 0000000..c617c4b --- /dev/null +++ b/web/src/main.rs @@ -0,0 +1,164 @@ +use std::collections::BTreeMap; +use axum::{ + Router, + routing::{get, service_method_routing as service}, + response::sse::{Event, KeepAlive, Sse}, +}; +use serde::Serialize; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::pin::Pin; +use std::task::{Context, Poll}; +use axum::error_handling::HandleErrorExt; +use axum::http::StatusCode; +use chrono::{DateTime, Duration, Utc}; +use futures::stream::Stream; +use rdkafka::Message; +use tokio::sync::mpsc::Receiver; +use tower_http::{services::ServeDir, trace::TraceLayer}; +use buska_core::config::KafkaClusterConfig; +use buska_core::search::bounds::{SearchBounds, SearchEnd, SearchStart}; +use buska_core::search::extractors::json::json_single_extract; +use buska_core::search::matchers::RegexMatch; +use buska_core::search::notifications::{PreparationStep, Progress, ProgressNotification, SearchNotification}; +use buska_core::search::SearchDefinition; +use buska_core::search_topic; + +#[tokio::main] +async fn main() { + + if std::env::var_os("RUST_LOG").is_none() { + std::env::set_var("RUST_LOG", "example_sse=debug,tower_http=debug") + } + + let static_files_service = + service::get(ServeDir::new("web/assets").append_index_html_on_directories(true)) + .handle_error(|error: std::io::Error| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Unhandled internal error: {}", error), + ) + }); + + let app = Router::new() + .fallback(static_files_service) + .route("/sse", get(sse_handler)) + .layer(TraceLayer::new_for_http()); + + // run it + let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); + axum::Server::bind(&addr) + .serve(app.into_make_service()) + .await + .unwrap(); +} + +async fn sse_handler() -> Sse>> { + let conf = KafkaClusterConfig { + bootstrap_servers: "localhost:9092".to_string(), + security: None + }; + let bounds = SearchBounds { + start: SearchStart::Earliest, + end: SearchEnd::CurrentLast + }; + let (sender, receiver) = tokio::sync::mpsc::channel::(1024); + let matcher = RegexMatch::new("some-[0-9]{1}$").unwrap(); + let extractor = json_single_extract("$.nested.string").unwrap(); + tokio::task::spawn(async move { + search_topic( + conf, + "test".to_string(), + sender, + bounds, + &mut SearchDefinition::new(extractor, Box::new(matcher)), + Duration::milliseconds(100) + ).await; + }); + Sse::new(SseSearchNotifications { recv: receiver }) + .keep_alive(KeepAlive::default()) +} + +struct SseSearchNotifications { + recv: Receiver, +} + +impl tokio_stream::Stream for SseSearchNotifications { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, co: &mut Context<'_>) -> Poll> { + self.recv + .poll_recv(co) + .map(|maybe_notif| maybe_notif.map(notif_as_event)) + } +} + +#[derive(Serialize)] +struct SearchSseEvent { + step: usize, + description: String +} + +#[derive(Serialize)] +struct ProgressSseEvent { + pub overall_progress: Progress, + pub per_partition_progress: BTreeMap, + pub matches: i64, + pub elapsed: i64, + pub eta: DateTime +} + +impl ProgressSseEvent { + fn new(progress: ProgressNotification) -> Self { + ProgressSseEvent { + overall_progress: progress.overall_progress, + per_partition_progress: string_keys(progress.per_partition_progress), + matches: progress.matches, + elapsed: progress.elapsed.num_milliseconds(), + eta: progress.eta + } + } +} + +fn string_keys(map: BTreeMap) -> BTreeMap { + map.into_iter().map(|(k, v)| (k.to_string(), v)).collect() +} + +fn notif_as_event(notif: SearchNotification) -> Result { + let e = match notif { + SearchNotification::Prepare(PreparationStep::CreateClient) => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 1, description: "Creating Kafka client".to_string() }).unwrap()), + SearchNotification::Prepare(PreparationStep::FetchMetadata) => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 2, description: "Fetching topic metadata".to_string() }).unwrap()), + SearchNotification::Prepare(PreparationStep::FetchWatermarks) => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 3, description: "Fetching topic watermarks (min/max)".to_string() }).unwrap()), + SearchNotification::Prepare(PreparationStep::SeekPartitions) => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 4, description: "Seeking partitions to desired offsets".to_string() }).unwrap()), + SearchNotification::Start => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 5, description: "Search started".to_string() }).unwrap()), + SearchNotification::Finish(_) => + Event::default() + .event("step") + .data(serde_json::to_string(&SearchSseEvent { step: 6, description: "Search finished".to_string() }).unwrap()), + SearchNotification::Progress(p) => + Event::default() + .event("search-progress") + .data(serde_json::to_string(&ProgressSseEvent::new(p)).unwrap()), + SearchNotification::Match(m) => { + Event::default() + .event("match") + .data(m.payload_view::().unwrap().unwrap()) + } + }; + Ok(e) +} \ No newline at end of file