Merge pull request #2947 from factcast/issue2882

#2882: PgStoreTelemetry
factcast · Jun 14, 2024 · 9c1f55e · 9c1f55e
2 parents 58c6bf3 + 4b7b228
commit 9c1f55e
Show file tree

Hide file tree

Showing 29 changed files with 970 additions and 9 deletions.
diff --git a/...-boot2/src/main/java/org/factcast/example/client/spring/boot2/hello/HelloWorldRunner.java b/...-boot2/src/main/java/org/factcast/example/client/spring/boot2/hello/HelloWorldRunner.java
@@ -22,6 +22,7 @@
 import org.factcast.core.Fact;
 import org.factcast.core.FactCast;
 import org.factcast.core.spec.FactSpec;
+import org.factcast.core.subscription.Subscription;
 import org.factcast.core.subscription.SubscriptionRequest;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
@@ -61,13 +62,33 @@ public void run(String... args) throws Exception {
     fc.subscribe(
             SubscriptionRequest.catchup(FactSpec.ns("users").type("UserCreated").version(3))
                 .fromScratch(),
-            element -> System.out.println(element))
+            System.out::println)
         .awaitCatchup();
 
     fc.subscribe(
             SubscriptionRequest.catchup(FactSpec.ns("users").type("UserCreated").version(1))
                 .fromScratch(),
-            element -> System.out.println(element))
+            System.out::println)
         .awaitCatchup();
+
+    // Follow subscription
+    Subscription followSub =
+        fc.subscribe(
+            SubscriptionRequest.follow(FactSpec.ns("users").type("UserCreated").version(3))
+                .fromScratch(),
+            System.out::println);
+
+    Fact anotherFact =
+        Fact.builder()
+            .ns("users")
+            .type("UserCreated")
+            .version(3)
+            .id(UUID.randomUUID())
+            .build(
+                "{\"firstName\":\"John\",\"lastName\":\"Wayne\",\"salutation\":\"Mr\",\"displayName\":\"JW\"}");
+    fc.publish(anotherFact);
+    System.out.println("published " + anotherFact);
+
+    followSub.awaitCatchup(5000).close();
   }
 }
diff --git a/factcast-examples/factcast-example-server-telemetry/pom.xml b/factcast-examples/factcast-example-server-telemetry/pom.xml
@@ -0,0 +1,115 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.factcast</groupId>
+    <artifactId>factcast-examples</artifactId>
+    <version>0.7.9-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+  <artifactId>factcast-example-server-telemetry</artifactId>
+  <properties>
+    <!-- Pin your own Spring Boot version. -->
+    <spring-boot.version>3.3.0</spring-boot.version>
+    <docker.image.prefix>factcast</docker.image.prefix>
+  </properties>
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-dependencies</artifactId>
+        <version>${spring-boot.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+      <dependency>
+        <groupId>org.factcast</groupId>
+        <artifactId>factcast-bom</artifactId>
+        <version>${project.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-web</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-actuator</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.factcast</groupId>
+      <artifactId>factcast-spring-boot-autoconfigure</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.factcast</groupId>
+      <artifactId>factcast-server-grpc</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.factcast</groupId>
+      <artifactId>factcast-store</artifactId>
+    </dependency>
+    <!-- you want to remove this -->
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>postgresql</artifactId>
+      <version>${testcontainers.version}</version>
+    </dependency>
+    <!-- /you want to remove this -->
+  </dependencies>
+  <build>
+    <finalName>factcast</finalName>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-deploy-plugin</artifactId>
+        <configuration>
+          <skip>true</skip>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>repackage</goal>
+            </goals>
+            <configuration>
+              <executable>true</executable>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>com.spotify</groupId>
+        <artifactId>docker-maven-plugin</artifactId>
+        <version>1.2.2</version>
+        <configuration>
+          <imageName>${docker.image.prefix}/${project.artifactId}</imageName>
+          <dockerDirectory>src/main/docker</dockerDirectory>
+          <resources>
+            <resource>
+              <targetPath>/</targetPath>
+              <directory>${project.build.directory}</directory>
+              <include>${project.build.finalName}.jar</include>
+            </resource>
+          </resources>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>
diff --git a/factcast-examples/factcast-example-server-telemetry/src/main/docker/Dockerfile b/factcast-examples/factcast-example-server-telemetry/src/main/docker/Dockerfile
@@ -0,0 +1,7 @@
+FROM frolvlad/alpine-oraclejdk8:slim
+MAINTAINER uwe.schaefer <factcast-auth-7234@codesmell.de>
+VOLUME /tmp
+COPY factcast.jar app.jar
+RUN sh -c 'touch /app.jar'
+ENV JAVA_OPTS=""
+ENTRYPOINT [ "sh", "-c", "java $JAVA_OPTS -Djava.security.egd=file:/dev/./urandom -jar /app.jar" ]
diff --git a/...ample-server-telemetry/src/main/java/org/factcast/example/server/ActuatorContributor.java b/...ample-server-telemetry/src/main/java/org/factcast/example/server/ActuatorContributor.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright © 2017-2024 factcast.org
+ *
+ * Licensed 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.
+ */
+package org.factcast.example.server;
+
+import java.util.concurrent.TimeUnit;
+import org.factcast.example.server.telemetry.MyTelemetryListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.info.Info;
+import org.springframework.boot.actuate.info.InfoContributor;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ActuatorContributor implements InfoContributor {
+
+  @Autowired MyTelemetryListener listener;
+
+  public void contribute(Info.Builder builder) {
+    builder.withDetail(
+        "followingSubscriptionsInfo",
+        listener.getFollowingSubscriptionsInfo().stream()
+            .map(i -> i.getMessage() + " following for " + i.getTime(TimeUnit.SECONDS) + "s")
+            .toList());
+  }
+}
diff --git a/...ver-telemetry/src/main/java/org/factcast/example/server/ExampleServerTelemetryConfig.java b/...ver-telemetry/src/main/java/org/factcast/example/server/ExampleServerTelemetryConfig.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright © 2017-2024 factcast.org
+ *
+ * Licensed 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.
+ */
+package org.factcast.example.server;
+
+import org.factcast.example.server.telemetry.MyTelemetryListener;
+import org.factcast.store.internal.telemetry.PgStoreTelemetry;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.web.SecurityFilterChain;
+
+@Configuration
+public class ExampleServerTelemetryConfig {
+
+  @Bean
+  public MyTelemetryListener catchupTelemetryListener(PgStoreTelemetry telemetry) {
+    return new MyTelemetryListener(telemetry);
+  }
+
+  @Bean
+  public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
+    return http.authorizeHttpRequests(x -> x.anyRequest().permitAll()).build();
+  }
+}
diff --git a/...erver-telemetry/src/main/java/org/factcast/example/server/ExampleServerWithTelemetry.java b/...erver-telemetry/src/main/java/org/factcast/example/server/ExampleServerWithTelemetry.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright © 2017-2020 factcast.org
+ *
+ * Licensed 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.
+ */
+package org.factcast.example.server;
+
+import lombok.extern.slf4j.Slf4j;
+import org.postgresql.Driver;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.testcontainers.containers.PostgreSQLContainer;
+
+/**
+ * Spring boot starter for running a factcast server.
+ *
+ * <p>This should contain a pgsql backend and grpc API frontend.
+ *
+ * @author uwe.schaefer@prisma-capacity.eu
+ */
+@SuppressWarnings("ALL")
+@SpringBootApplication
+@Slf4j
+public class ExampleServerWithTelemetry {
+
+  public static void main(String[] args) {
+    // you will want to replace this with RDS, a local pgsql database or
+    // whatever your usecase is.
+    startPostgresContainer();
+
+    SpringApplication.run(ExampleServerWithTelemetry.class, args);
+  }
+
+  private static void startPostgresContainer() {
+    log.info("Trying to start postgres testcontainer");
+    PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:15.2");
+    postgres.start();
+    String url = postgres.getJdbcUrl();
+    System.setProperty("spring.datasource.driver-class-name", Driver.class.getName());
+    System.setProperty("spring.datasource.url", url);
+    System.setProperty("spring.datasource.username", postgres.getUsername());
+    System.setProperty("spring.datasource.password", postgres.getPassword());
+  }
+}
diff --git a/...er-telemetry/src/main/java/org/factcast/example/server/telemetry/MyTelemetryListener.java b/...er-telemetry/src/main/java/org/factcast/example/server/telemetry/MyTelemetryListener.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright © 2017-2024 factcast.org
+ *
+ * Licensed 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.
+ */
+package org.factcast.example.server.telemetry;
+
+import com.google.common.collect.Lists;
+import com.google.common.eventbus.Subscribe;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
+import org.factcast.store.internal.telemetry.PgStoreTelemetry;
+
+@RequiredArgsConstructor
+@Slf4j
+public class MyTelemetryListener {
+
+  final List<StopWatch> followingSubscriptionsInfo = Lists.newArrayList();
+
+  public MyTelemetryListener(PgStoreTelemetry telemetry) {
+    telemetry.register(this);
+  }
+
+  @Subscribe
+  public void on(PgStoreTelemetry.Connect signal) {
+    log.info("FactStreamTelemetry Connect: {}", signal.request());
+  }
+
+  @Subscribe
+  public void on(PgStoreTelemetry.Catchup signal) {
+    log.info("FactStreamTelemetry Catchup: {}", signal.request());
+  }
+
+  @Subscribe
+  public void on(PgStoreTelemetry.Follow signal) {
+    log.info("FactStreamTelemetry Follow: {}", signal.request());
+    StopWatch stopWatch = new StopWatch(signal.request().debugInfo());
+    stopWatch.start();
+    followingSubscriptionsInfo.add(stopWatch);
+  }
+
+  @Subscribe
+  public void on(PgStoreTelemetry.Complete signal) {
+    log.info("FactStreamTelemetry Complete: {}", signal.request());
+  }
+
+  @Subscribe
+  public void on(PgStoreTelemetry.Close signal) {
+    log.info("FactStreamTelemetry Close: {}", signal.request());
+    followingSubscriptionsInfo.removeIf(i -> i.getMessage().equals(signal.request().debugInfo()));
+  }
+
+  public List<StopWatch> getFollowingSubscriptionsInfo() {
+    return followingSubscriptionsInfo;
+  }
+}
diff --git a/...cast-examples/factcast-example-server-telemetry/src/main/resources/application.properties b/...cast-examples/factcast-example-server-telemetry/src/main/resources/application.properties
@@ -0,0 +1,15 @@
+spring.datasource.tomcat.max-wait=20000
+spring.datasource.tomcat.remove-abandoned-timeout=360000
+spring.datasource.tomcat.test-on-borrow=true
+# ensure RDS reconnect behavior (issue #53)
+spring.datasource.tomcat.connectionProperties=socketTimeout=20;connectTimeout=10;loginTimeout=10;
+factcast.security.enabled=false
+#spring.datasource.url=jdbc:postgresql://localhost/factcast
+#spring.datasource.username=doc
+#spring.datasource.password=doc
+factcast.store.schemaRegistryUrl=classpath:example-registry
+#factcast.store.allowUnvalidatedPublish=true
+#factcast.store.persistentSchemaStore=false
+management.server.port=9091
+management.endpoints.web.exposure.include=*
+management.endpoint.info.enabled=true
diff --git a/factcast-examples/factcast-example-server-telemetry/src/main/resources/banner.txt b/factcast-examples/factcast-example-server-telemetry/src/main/resources/banner.txt
@@ -0,0 +1,10 @@
+
+
+  ███████╗ █████╗  ██████╗████████╗ ██████╗ █████╗ ███████╗████████╗
+  ██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔════╝██╔══██╗██╔════╝╚══██╔══╝
+  █████╗  ███████║██║        ██║   ██║     ███████║███████╗   ██║   
+  ██╔══╝  ██╔══██║██║        ██║   ██║     ██╔══██║╚════██║   ██║   
+  ██║     ██║  ██║╚██████╗   ██║   ╚██████╗██║  ██║███████║   ██║   
+  ╚═╝     ╚═╝  ╚═╝ ╚═════╝   ╚═╝    ╚═════╝╚═╝  ╚═╝╚══════╝   ╚═╝   
+  factcast-example-server-telemetry
+
diff --git a/...examples/factcast-example-server-telemetry/src/main/resources/example-registry/index.json b/...examples/factcast-example-server-telemetry/src/main/resources/example-registry/index.json
@@ -0,0 +1 @@
+{"schemes":[{"id":"users/UserCreated/1/schema.json","ns":"users","type":"UserCreated","version":1,"hash":"24d48268356e3cb7ac2f148850e4aac1"},{"id":"users/UserCreated/2/schema.json","ns":"users","type":"UserCreated","version":2,"hash":"8fb5294baa827a46f5f0138fe08f04f3"},{"id":"users/UserCreated/3/schema.json","ns":"users","type":"UserCreated","version":3,"hash":"f61380a21bcd935a3a1b919b90374d61"}],"transformations":[{"id":"users/UserCreated/1-2/transform.js","ns":"users","type":"UserCreated","from":1,"to":2,"hash":"546d80170d224d6f24b276f3b8c4656e"},{"id":"users/UserCreated/3-2/transform.js","ns":"users","type":"UserCreated","from":3,"to":2,"hash":"e32be57fbe84391b9415b49cf730d400"},{"id":"users/UserCreated/2-3/transform.js","ns":"users","type":"UserCreated","from":2,"to":3,"hash":"bf85130f22ed18a0b559dc904059457d"},{"ns":"users","type":"UserCreated","from":2,"to":1,"id":"synthetic/users/UserCreated/2-1/transform.js"}]}