fix: sample improvements

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java

@@ -81,6 +81,16 @@ private void pollForResource(R resource) {
     }
   }
 
+  private Optional<T> getAndCacheResource(ResourceID resourceID) {
+    var resource = resourceCache.get(resourceID);
+    if (resource.isPresent()) {
+      var value = resourceSupplier.getResources(resource.get());
+      value.ifPresent(v -> cache.put(resourceID, v));
+      return value;
+    }
+    return Optional.empty();
+  }
+
   @Override
   public void onResourceCreated(R resource) {
     checkAndRegisterTask(resource);
@@ -121,6 +131,22 @@ public void run() {
     }
   }
 
+  /**
+   *
+   * @param resourceID of the target related resource
+   * @return the cached value of the resource, if not present it gets the resource from the
+   *         supplier. If the supplier provides a value it is cached, so there will be no new event
+   *         related for the new event.
+   */
+  public Optional<T> getValueFromCacheOrSupplier(ResourceID resourceID) {
+    var cachedValue = getCachedValue(resourceID);
+    if (cachedValue.isPresent()) {
+      return cachedValue;
+    } else {
+      return getAndCacheResource(resourceID);
+    }
+  }
+
   public interface ResourceSupplier<T, R> {
     Optional<T> getResources(R resource);
   }

sample-operators/mysql-schema/pom.xml

@@ -67,6 +67,19 @@
             <artifactId>jackson-dataformat-yaml</artifactId>
             <version>2.13.0</version>
         </dependency>
+        <dependency>
+            <groupId>javax.cache</groupId>
+            <artifactId>cache-api</artifactId>
+            <version>${jcache.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>jcache</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java

@@ -6,6 +6,11 @@
 import java.util.Base64;
 import java.util.Optional;
 
+import javax.cache.Cache;
+import javax.cache.CacheManager;
+import javax.cache.configuration.MutableConfiguration;
+import javax.cache.spi.CachingProvider;
+
 import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -15,30 +20,55 @@
 import io.fabric8.kubernetes.api.model.SecretBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
+import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource;
+import io.javaoperatorsdk.operator.sample.schema.Schema;
 import io.javaoperatorsdk.operator.sample.schema.SchemaService;
 
+import com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider;
+
 import static java.lang.String.format;
 
 @ControllerConfiguration
 public class MySQLSchemaReconciler
-    implements Reconciler<MySQLSchema>, ErrorStatusHandler<MySQLSchema> {
+    implements Reconciler<MySQLSchema>, ErrorStatusHandler<MySQLSchema>,
+    EventSourceInitializer<MySQLSchema> {
   public static final String SECRET_FORMAT = "%s-secret";
   public static final String USERNAME_FORMAT = "%s-user";
+  public static final int POLL_PERIOD = 500;
   private final Logger log = LoggerFactory.getLogger(getClass());
 
   private final KubernetesClient kubernetesClient;
   private final MySQLDbConfig mysqlDbConfig;
+  PerResourcePollingEventSource<Schema, MySQLSchema> perResourcePollingEventSource;
 
   public MySQLSchemaReconciler(KubernetesClient kubernetesClient, MySQLDbConfig mysqlDbConfig) {
     this.kubernetesClient = kubernetesClient;
     this.mysqlDbConfig = mysqlDbConfig;
   }
 
+  @Override
+  public void prepareEventSources(EventSourceRegistry<MySQLSchema> eventSourceRegistry) {
+    CachingProvider cachingProvider = new CaffeineCachingProvider();
+    CacheManager cacheManager = cachingProvider.getCacheManager();
+    Cache<ResourceID, Schema> schemaCache =
+        cacheManager.createCache("schema-cache", new MutableConfiguration<>());
+
+    perResourcePollingEventSource =
+        new PerResourcePollingEventSource<>(new SchemaPollingResourceSupplier(mysqlDbConfig),
+            eventSourceRegistry.getControllerResourceEventSource().getResourceCache(), POLL_PERIOD,
+            schemaCache);
+
+    eventSourceRegistry.registerEventSource(perResourcePollingEventSource);
+  }
+
   @Override
   public UpdateControl<MySQLSchema> reconcile(MySQLSchema schema,
       Context context) {
+    var dbSchema = perResourcePollingEventSource
+        .getValueFromCacheOrSupplier(ResourceID.fromResource(schema));
     try (Connection connection = getConnection()) {
-      var dbSchema = SchemaService.getSchema(connection, schema.getMetadata().getName());
       if (!dbSchema.isPresent()) {
         var schemaName = schema.getMetadata().getName();
         String password = RandomStringUtils.randomAlphanumeric(16);
@@ -115,6 +145,12 @@ private void updateStatusPojo(MySQLSchema schema, String secretName, String user
 
   private void createSecret(MySQLSchema schema, String password, String secretName,
       String userName) {
+
+    var currentSecret = kubernetesClient.secrets().inNamespace(schema.getMetadata().getNamespace())
+        .withName(secretName).get();
+    if (currentSecret != null) {
+      return;
+    }
     Secret credentialsSecret =
         new SecretBuilder()
             .withNewMetadata()
@@ -133,4 +169,6 @@ private void createSecret(MySQLSchema schema, String password, String secretName
         .inNamespace(schema.getMetadata().getNamespace())
         .create(credentialsSecret);
   }
+
+
 }

sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaPollingResourceSupplier.java

@@ -0,0 +1,22 @@
+package io.javaoperatorsdk.operator.sample;
+
+import java.util.Optional;
+
+import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource;
+import io.javaoperatorsdk.operator.sample.schema.Schema;
+import io.javaoperatorsdk.operator.sample.schema.SchemaService;
+
+public class SchemaPollingResourceSupplier
+    implements PerResourcePollingEventSource.ResourceSupplier<Schema, MySQLSchema> {
+
+  private final SchemaService schemaService;
+
+  public SchemaPollingResourceSupplier(MySQLDbConfig mySQLDbConfig) {
+    this.schemaService = new SchemaService(mySQLDbConfig);
+  }
+
+  @Override
+  public Optional<Schema> getResources(MySQLSchema resource) {
+    return schemaService.getSchema(resource.getMetadata().getName());
+  }
+}

sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaStatus.java

@@ -1,6 +1,8 @@
 package io.javaoperatorsdk.operator.sample;
 
-public class SchemaStatus {
+import io.javaoperatorsdk.operator.api.ObservedGenerationAwareStatus;
+
+public class SchemaStatus extends ObservedGenerationAwareStatus {
 
   private String url;
 

sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/Schema.java

@@ -1,8 +1,9 @@
 package io.javaoperatorsdk.operator.sample.schema;
 
+import java.io.Serializable;
 import java.util.Objects;
 
-public class Schema {
+public class Schema implements Serializable {
 
   private String name;
   private String characterSet;

sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/SchemaService.java

@@ -40,11 +40,11 @@ public static void createSchemaAndRelatedUser(Connection connection, String sche
                 "CREATE SCHEMA `%1$s` DEFAULT CHARACTER SET %2$s",
                 schemaName, encoding));
       }
-
-      try (Statement statement = connection.createStatement()) {
-        statement.execute(format("CREATE USER '%1$s' IDENTIFIED BY '%2$s'", userName, password));
+      if (!userExists(connection, userName)) {
+        try (Statement statement = connection.createStatement()) {
+          statement.execute(format("CREATE USER '%1$s' IDENTIFIED BY '%2$s'", userName, password));
+        }
       }
-
       try (Statement statement = connection.createStatement()) {
         statement.execute(
             format("GRANT ALL ON `%1$s`.* TO '%2$s'", schemaName, userName));
@@ -73,6 +73,19 @@ public static void deleteSchemaAndRelatedUser(Connection connection, String sche
     }
   }
 
+  private static boolean userExists(Connection connection, String username) {
+    try (PreparedStatement ps =
+        connection.prepareStatement(
+            "SELECT EXISTS(SELECT 1 FROM mysql.user WHERE user = ?)")) {
+      ps.setString(1, username);
+      try (ResultSet resultSet = ps.executeQuery()) {
+        return resultSet.next();
+      }
+    } catch (SQLException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
   public static Optional<Schema> getSchema(Connection connection, String schemaName) {
     try (PreparedStatement ps =
         connection.prepareStatement(