feature: add jaeger opentracing improvements

Feature/mongo

pom.xml

@@ -17,6 +17,26 @@
 		<java.version>17</java.version>
 	</properties>
 	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-sleuth-zipkin</artifactId>
+			<version>3.1.1</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-sleuth</artifactId>
+			<version>3.1.1</version>
+		</dependency>
+		<dependency>
+			<groupId>io.micrometer</groupId>
+			<artifactId>micrometer-registry-prometheus</artifactId>
+			<version>1.8.4</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springdoc</groupId>
+			<artifactId>springdoc-openapi-ui</artifactId>
+			<version>1.6.7</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-data-mongodb</artifactId>

src/main/java/com/eventsourcing/bankAccount/commands/BankAccountCommandHandler.java

@@ -5,17 +5,20 @@ import com.eventsourcing.bankAccount.domain.BankAccountAggregate;
 import com.eventsourcing.es.EventStoreDB;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cloud.sleuth.annotation.NewSpan;
+import org.springframework.cloud.sleuth.annotation.SpanTag;
 import org.springframework.stereotype.Service;
 
 @RequiredArgsConstructor
 @Slf4j
 @Service
-public class BankAccountCommandHandler implements BankAccountCommandService{
+public class BankAccountCommandHandler implements BankAccountCommandService {
 
     private final EventStoreDB eventStoreDB;
 
     @Override
-    public String handle(CreateBankAccountCommand command) {
+    @NewSpan
+    public String handle(@SpanTag("command") CreateBankAccountCommand command) {
         final var aggregate = new BankAccountAggregate(command.aggregateID());
         aggregate.createBankAccount(command.email(), command.address(), command.userName());
         eventStoreDB.save(aggregate);
@@ -25,7 +28,8 @@ public class BankAccountCommandHandler implements BankAccountCommandService{
     }
 
     @Override
-    public void handle(ChangeEmailCommand command) {
+    @NewSpan
+    public void handle(@SpanTag("command") ChangeEmailCommand command) {
         final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
         aggregate.changeEmail(command.newEmail());
         eventStoreDB.save(aggregate);
@@ -33,7 +37,8 @@ public class BankAccountCommandHandler implements BankAccountCommandService{
     }
 
     @Override
-    public void handle(ChangeAddressCommand command) {
+    @NewSpan
+    public void handle(@SpanTag("command") ChangeAddressCommand command) {
         final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
         aggregate.changeAddress(command.newAddress());
         eventStoreDB.save(aggregate);
@@ -41,7 +46,8 @@ public class BankAccountCommandHandler implements BankAccountCommandService{
     }
 
     @Override
-    public void handle(DepositAmountCommand command) {
+    @NewSpan
+    public void handle(@SpanTag("command") DepositAmountCommand command) {
         final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
         aggregate.depositBalance(command.amount());
         eventStoreDB.save(aggregate);

src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java

@@ -15,6 +15,8 @@ import com.eventsourcing.es.SerializerUtils;
 import com.eventsourcing.mappers.BankAccountMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cloud.sleuth.annotation.NewSpan;
+import org.springframework.cloud.sleuth.annotation.SpanTag;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
 import org.springframework.kafka.support.Acknowledgment;
@@ -54,7 +56,8 @@ public class BankAccountMongoProjection implements Projection {
         }
     }
 
-    private void processEvents(List<Event> events) {
+    @NewSpan
+    private void processEvents(@SpanTag("events") List<Event> events) {
         if (events.isEmpty()) return;
 
         try {
@@ -69,7 +72,8 @@ public class BankAccountMongoProjection implements Projection {
     }
 
     @Override
-    public void when(Event event) {
+    @NewSpan
+    public void when(@SpanTag("event") Event event) {
         final var aggregateId = event.getAggregateId();
         log.info("(when) >>>>> aggregateId: {}", aggregateId);
 
@@ -87,7 +91,8 @@ public class BankAccountMongoProjection implements Projection {
     }
 
 
-    private void handle(BankAccountCreatedEvent event) {
+    @NewSpan
+    private void handle(@SpanTag("event") BankAccountCreatedEvent event) {
         log.info("(when) BankAccountCreatedEvent: {}, aggregateID: {}", event, event.getAggregateId());
 
         final var document = BankAccountDocument.builder()
@@ -102,7 +107,8 @@ public class BankAccountMongoProjection implements Projection {
         log.info("(BankAccountCreatedEvent) insert: {}", insert);
     }
 
-    private void handle(EmailChangedEvent event) {
+    @NewSpan
+    private void handle(@SpanTag("event") EmailChangedEvent event) {
         log.info("(when) EmailChangedEvent: {}, aggregateID: {}", event, event.getAggregateId());
         Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
         if (documentOptional.isEmpty())
@@ -113,7 +119,8 @@ public class BankAccountMongoProjection implements Projection {
         mongoRepository.save(document);
     }
 
-    private void handle(AddressUpdatedEvent event) {
+    @NewSpan
+    private void handle(@SpanTag("event") AddressUpdatedEvent event) {
         log.info("(when) AddressUpdatedEvent: {}, aggregateID: {}", event, event.getAggregateId());
         Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
         if (documentOptional.isEmpty())
@@ -124,7 +131,8 @@ public class BankAccountMongoProjection implements Projection {
         mongoRepository.save(document);
     }
 
-    private void handle(BalanceDepositedEvent event) {
+    @NewSpan
+    private void handle(@SpanTag("event") BalanceDepositedEvent event) {
         log.info("(when) BalanceDepositedEvent: {}, aggregateID: {}", event, event.getAggregateId());
         Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
         if (documentOptional.isEmpty())

src/main/java/com/eventsourcing/bankAccount/queries/BankAccountQueryHandler.java

@@ -8,6 +8,8 @@ import com.eventsourcing.es.EventStoreDB;
 import com.eventsourcing.mappers.BankAccountMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cloud.sleuth.annotation.NewSpan;
+import org.springframework.cloud.sleuth.annotation.SpanTag;
 import org.springframework.stereotype.Service;
 
 import java.util.Optional;
@@ -22,7 +24,8 @@ public class BankAccountQueryHandler implements BankAccountQueryService {
     private final BankAccountMongoRepository mongoRepository;
 
     @Override
-    public BankAccountResponseDTO handle(GetBankAccountByIDQuery query) {
+    @NewSpan
+    public BankAccountResponseDTO handle(@SpanTag("query") GetBankAccountByIDQuery query) {
         Optional<BankAccountDocument> optionalDocument = mongoRepository.findByAggregateId(query.aggregateID());
         if (optionalDocument.isPresent()) {
             return BankAccountMapper.bankAccountResponseDTOFromDocument(optionalDocument.get());

src/main/java/com/eventsourcing/configuration/SwaggerOpenAPIConfiguration.java

@@ -0,0 +1,15 @@
+package com.eventsourcing.configuration;
+
+
+import io.swagger.v3.oas.annotations.OpenAPIDefinition;
+import io.swagger.v3.oas.annotations.info.Contact;
+import io.swagger.v3.oas.annotations.info.Info;
+import org.springframework.stereotype.Component;
+
+
+@OpenAPIDefinition(info = @Info(title = "Spring CQRS and Event Sourcing Microservice",
+        description = "Spring Postgresql MongoDB Kafka CQRS and Event Sourcing Microservice",
+        contact = @Contact(name = "Alexander Bryksin", email = "alexander.bryksin@yandex.ru", url = "https://github.com/AleksK1NG")))
+@Component
+public class SwaggerOpenAPIConfiguration {
+}

src/main/java/com/eventsourcing/es/EventStore.java

@@ -4,6 +4,8 @@ package com.eventsourcing.es;
 import com.eventsourcing.es.exceptions.AggregateNotFoundException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cloud.sleuth.annotation.NewSpan;
+import org.springframework.cloud.sleuth.annotation.SpanTag;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Repository;
@@ -28,7 +30,8 @@ public class EventStore implements EventStoreDB {
     private final EventBus eventBus;
 
     @Override
-    public void saveEvents(List<Event> events) {
+    @NewSpan
+    public void saveEvents(@SpanTag("events") List<Event> events) {
         if (events.isEmpty()) return;
 
         final List<Event> changes = new ArrayList<>(events);
@@ -46,7 +49,8 @@ public class EventStore implements EventStoreDB {
     }
 
     @Override
-    public List<Event> loadEvents(String aggregateId, long version) {
+    @NewSpan
+    public List<Event> loadEvents(@SpanTag("aggregateId") String aggregateId, @SpanTag("version") long version) {
         final List<Event> events = jdbcTemplate.query(LOAD_EVENTS_QUERY, Map.of("aggregate_id", aggregateId, "version", version),
                 (rs, rowNum) -> Event.builder()
                         .aggregateId(rs.getString("aggregate_id"))
@@ -62,7 +66,8 @@ public class EventStore implements EventStoreDB {
         return events;
     }
 
-    private <T extends AggregateRoot> void saveSnapshot(T aggregate) {
+    @NewSpan
+    private <T extends AggregateRoot> void saveSnapshot(@SpanTag("aggregate") T aggregate) {
         aggregate.toSnapshot();
         final var snapshot = EventSourcingUtils.snapshotFromAggregate(aggregate);
 
@@ -78,7 +83,8 @@ public class EventStore implements EventStoreDB {
 
     @Override
     @Transactional
-    public <T extends AggregateRoot> void save(T aggregate) {
+    @NewSpan
+    public <T extends AggregateRoot> void save(@SpanTag("aggregate") T aggregate) {
         final List<Event> aggregateEvents = new ArrayList<>(aggregate.getChanges());
 
         if (aggregate.getVersion() > 1) {
@@ -95,7 +101,8 @@ public class EventStore implements EventStoreDB {
         log.info("(save) saved aggregate: {}", aggregate);
     }
 
-    private void handleConcurrency(String aggregateId) {
+    @NewSpan
+    private void handleConcurrency(@SpanTag("aggregateId") String aggregateId) {
         try {
             String aggregateID = jdbcTemplate.queryForObject(HANDLE_CONCURRENCY_QUERY, Map.of("aggregate_id", aggregateId), String.class);
             log.info("(handleConcurrency) aggregateID for lock: {}", aggregateID);
@@ -105,7 +112,8 @@ public class EventStore implements EventStoreDB {
         log.info("(handleConcurrency) aggregateID for lock: {}", aggregateId);
     }
 
-    private Optional<Snapshot> loadSnapshot(String aggregateId) {
+    @NewSpan
+    private Optional<Snapshot> loadSnapshot(@SpanTag("aggregateId") String aggregateId) {
         final Optional<Snapshot> snapshot = jdbcTemplate.query(LOAD_SNAPSHOT_QUERY, Map.of("aggregate_id", aggregateId), (rs, rowNum) -> Snapshot.builder()
                 .aggregateId(rs.getString("aggregate_id"))
                 .aggregateType(rs.getString("aggregate_type"))
@@ -119,7 +127,8 @@ public class EventStore implements EventStoreDB {
         return snapshot;
     }
 
-    private <T extends AggregateRoot> T getAggregate(final String aggregateId, final Class<T> aggregateType) {
+    @NewSpan
+    private <T extends AggregateRoot> T getAggregate(@SpanTag("aggregateId") final String aggregateId, @SpanTag("aggregateType") final Class<T> aggregateType) {
         try {
             return aggregateType.getConstructor(String.class).newInstance(aggregateId);
         } catch (Exception ex) {
@@ -127,7 +136,8 @@ public class EventStore implements EventStoreDB {
         }
     }
 
-    private <T extends AggregateRoot> T getSnapshotFromClass(Optional<Snapshot> snapshot, String aggregateId, Class<T> aggregateType) {
+    @NewSpan
+    private <T extends AggregateRoot> T getSnapshotFromClass(@SpanTag("snapshot") Optional<Snapshot> snapshot, @SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
         if (snapshot.isEmpty()) {
             final var defaultSnapshot = EventSourcingUtils.snapshotFromAggregate(getAggregate(aggregateId, aggregateType));
             return EventSourcingUtils.aggregateFromSnapshot(defaultSnapshot, aggregateType);
@@ -137,7 +147,8 @@ public class EventStore implements EventStoreDB {
 
     @Override
     @Transactional(readOnly = true)
-    public <T extends AggregateRoot> T load(String aggregateId, Class<T> aggregateType) {
+    @NewSpan
+    public <T extends AggregateRoot> T load(@SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
 
         final Optional<Snapshot> snapshot = this.loadSnapshot(aggregateId);
 
@@ -156,7 +167,8 @@ public class EventStore implements EventStoreDB {
     }
 
     @Override
-    public Boolean exists(String aggregateId) {
+    @NewSpan
+    public Boolean exists(@SpanTag("aggregateId") String aggregateId) {
         try {
             final var id = jdbcTemplate.queryForObject(EXISTS_QUERY, Map.of("aggregate_id", aggregateId), String.class);
             log.info("aggregate exists id: {}", id);

src/main/java/com/eventsourcing/es/KafkaEventBus.java

@@ -5,6 +5,8 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.sleuth.annotation.NewSpan;
+import org.springframework.cloud.sleuth.annotation.SpanTag;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 
@@ -21,7 +23,8 @@ public class KafkaEventBus implements EventBus {
     private String bankAccountTopicName;
 
     @Override
-    public void publish(List<Event> events) {
+    @NewSpan
+    public void publish(@SpanTag("events") List<Event> events) {
         final byte[] eventsBytes = SerializerUtils.serializeToJsonBytes(events.toArray(new Event[]{}));
         final ProducerRecord<String, byte[]> record = new ProducerRecord<>(bankAccountTopicName, eventsBytes);
 

src/main/resources/application.properties

@@ -36,3 +36,11 @@ spring.data.mongodb.authentication-database=admin
 spring.data.mongodb.username=admin
 spring.data.mongodb.password=admin
 spring.data.mongodb.database=microservices
+
+springdoc.swagger-ui.path=/swagger-ui.html
+
+management.endpoints.web.exposure.include=health,prometheus,info
+
+spring.sleuth.propagation.type=w3c,b3
+spring.sleuth.opentracing.enabled=true
+spring.zipkin.base-url=http://localhost:9411