Replace Redis with Apache Ignite for in memory cache and db (futurewe…

…i-cloud#129)

* Add ignite cache

* Add db modue

* Fix test case failure

* Add transaction for db cahche and SSL for ignite connection

* SSL for ignite connection is optional

* Set the default value of ignite's SSL configuration to null

* Rename the directory cache.repo to db.repo

pom.xml

@@ -130,6 +130,12 @@
             <artifactId>spring-kafka-test</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>2.8.0</version>
+        </dependency>
     </dependencies>
 
     <build>

src/com/futurewei/alcor/controller/app/AlcorControllerApp.java

@@ -18,7 +18,7 @@
 
 import com.futurewei.alcor.controller.app.onebox.OneBoxConfig;
 import com.futurewei.alcor.controller.app.onebox.OneBoxUtil;
-import com.futurewei.alcor.controller.cache.config.RedisConfiguration;
+import com.futurewei.alcor.controller.db.redis.RedisConfiguration;
 import com.futurewei.alcor.controller.logging.Logger;
 import com.futurewei.alcor.controller.logging.LoggerFactory;
 import com.futurewei.alcor.controller.model.HostInfo;

src/com/futurewei/alcor/controller/db/CacheFactory.java

@@ -0,0 +1,70 @@
+/*
+Copyright 2019 The Alcor Authors.
+
+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 com.futurewei.alcor.controller.db;
+
+import com.futurewei.alcor.controller.db.ignite.IgniteCache;
+import com.futurewei.alcor.controller.db.redis.RedisCache;
+import org.apache.ignite.client.IgniteClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.stereotype.Component;
+
+
+@ComponentScan
+@Component
+public class CacheFactory {
+    @Autowired(required = false)
+    private IgniteClient igniteClient;
+
+    @Autowired
+    LettuceConnectionFactory lettuceConnectionFactory;
+
+    @Bean
+    CacheFactory cacheFactoryInstance() {
+        return new CacheFactory();
+    }
+
+    private ICache getIgniteCache(String cacheName) {
+        return new IgniteCache<>(igniteClient, cacheName);
+    }
+
+    public <K, V>ICache getRedisCache(Class<V> v, String cacheName) {
+
+        RedisTemplate<K, V> template = new RedisTemplate<K, V>();
+        template.setConnectionFactory(lettuceConnectionFactory);
+        template.setKeySerializer(new StringRedisSerializer());
+
+        template.setHashValueSerializer(new Jackson2JsonRedisSerializer<>(v));
+        template.setValueSerializer(new Jackson2JsonRedisSerializer<>(v));
+        template.afterPropertiesSet();
+
+        return new RedisCache<>(template, cacheName);
+    }
+
+    public <K, V>ICache getCache(Class<V> v) {
+        if (igniteClient != null) {
+            return getIgniteCache(v.getName());
+        }
+
+        return getRedisCache(v, v.getName());
+    }
+}

src/com/futurewei/alcor/controller/db/ICache.java

@@ -0,0 +1,37 @@
+/*
+Copyright 2019 The Alcor Authors.
+
+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 com.futurewei.alcor.controller.db;
+
+import com.futurewei.alcor.controller.exception.CacheException;
+
+import java.util.Map;
+
+public interface ICache<K, V> {
+    V get(K var1) throws CacheException;
+
+    void put(K var1, V var2) throws CacheException;
+
+    boolean containsKey(K var1) throws CacheException;
+
+    Map<K, V> getAll() throws CacheException;
+
+    void putAll(Map<? extends K, ? extends V> var1) throws CacheException;
+
+    boolean remove(K var1) throws CacheException;
+
+    Transaction getTransaction();
+}

src/com/futurewei/alcor/controller/cache/repo/ICacheRepository.java → src/com/futurewei/alcor/controller/db/Transaction.java

@@ -14,17 +14,16 @@
         limitations under the License.
 */
 
-package com.futurewei.alcor.controller.cache.repo;
+package com.futurewei.alcor.controller.db;
 
-import java.util.Map;
+import com.futurewei.alcor.controller.exception.CacheException;
 
-public interface ICacheRepository<T> {
+public interface Transaction {
+    void start() throws CacheException;
 
-    T findItem(String id);
+    void commit() throws CacheException;
 
-    Map<String, T> findAllItems();
+    void rollback() throws CacheException;
 
-    void addItem(T newItem);
-
-    void deleteItem(String id);
+    void close();
 }

src/com/futurewei/alcor/controller/db/ignite/IgniteCache.java

@@ -0,0 +1,130 @@
+/*
+Copyright 2019 The Alcor Authors.
+
+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 com.futurewei.alcor.controller.db.ignite;
+
+import com.futurewei.alcor.controller.db.ICache;
+import com.futurewei.alcor.controller.db.Transaction;
+import com.futurewei.alcor.controller.exception.CacheException;
+import com.futurewei.alcor.controller.logging.Logger;
+import com.futurewei.alcor.controller.logging.LoggerFactory;
+import org.apache.ignite.cache.query.Query;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.client.ClientCache;
+import org.apache.ignite.client.ClientException;
+import org.apache.ignite.client.IgniteClient;
+import org.springframework.util.Assert;
+
+import javax.cache.Cache;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.stream.Collectors;
+
+
+public class IgniteCache<K, V> implements ICache<K, V> {
+    private static final Logger logger = LoggerFactory.getLogger();
+    private ClientCache<K, V> cache;
+    private IgniteClient igniteClient;
+    private IgniteTransaction transaction;
+
+    public IgniteCache(IgniteClient igniteClient, String name) {
+        this.igniteClient = igniteClient;
+
+        try {
+            cache = igniteClient.getOrCreateCache(name);
+        }
+        catch (ClientException e) {
+            logger.log(Level.WARNING, "Create cache for vpc failed:" + e.getMessage());
+        }
+        catch (Exception e) {
+            logger.log(Level.WARNING, "Unexpected failure:" + e.getMessage());
+        }
+
+        transaction = new IgniteTransaction(igniteClient);
+
+        Assert.notNull(igniteClient, "Create cache for vpc failed");
+    }
+
+    @Override
+    public V get(K key) throws CacheException {
+        try {
+            return cache.get(key);
+        }catch (ClientException e) {
+            logger.log(Level.WARNING, "IgniteCache get operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public void put(K key, V value) throws CacheException {
+        try {
+            cache.put(key, value);
+        }catch (ClientException e) {
+            logger.log(Level.WARNING, "IgniteCache put operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public boolean containsKey(K key) throws CacheException {
+        try {
+            return cache.containsKey(key);
+        }catch (ClientException e) {
+            logger.log(Level.WARNING, "IgniteCache containsKey operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public Map<K, V> getAll() throws CacheException {
+        Query<Cache.Entry<K, V>> qry = new ScanQuery<K, V>();
+
+        try {
+            QueryCursor<Cache.Entry<K, V>> cur = cache.query(qry);
+            return cur.getAll().stream().collect(Collectors
+                    .toMap(Cache.Entry::getKey, Cache.Entry::getValue));
+        } catch (Exception e) {
+            logger.log(Level.WARNING, "IgniteCache getAll operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public void putAll(Map<? extends K, ? extends V> items) throws CacheException {
+        try {
+            cache.putAll(items);
+        }catch (ClientException e) {
+            logger.log(Level.WARNING, "IgniteCache putAll operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public boolean remove(K key) throws CacheException {
+        try {
+            return cache.remove(key);
+        }catch (ClientException e) {
+            logger.log(Level.WARNING, "IgniteCache remove operation error:" + e.getMessage());
+            throw new CacheException(e.getMessage());
+        }
+    }
+
+    @Override
+    public Transaction getTransaction() {
+        return transaction;
+    }
+}

src/com/futurewei/alcor/controller/db/ignite/IgniteConfiguration.java

@@ -0,0 +1,88 @@
+/*
+Copyright 2019 The Alcor Authors.
+
+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 com.futurewei.alcor.controller.db.ignite;
+
+import com.futurewei.alcor.controller.logging.Logger;
+import com.futurewei.alcor.controller.logging.LoggerFactory;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.util.Assert;
+import java.util.logging.Level;
+
+@Configuration
+@ComponentScan("com.futurewei.alcor.controller.db")
+@EntityScan("com.futurewei.alcor.controller.db")
+@ConditionalOnProperty(prefix = "ignite", name = "host")
+public class IgniteConfiguration {
+    private static final Logger logger = LoggerFactory.getLogger();
+
+    @Value("${ignite.host}")
+    private String host;
+
+    @Value("${ignite.port}")
+    private Integer port;
+
+    @Value("${ignite.key-store-path:#{null}}")
+    private String keyStorePath;
+
+    @Value("${ignite.key-store-password:#{null}}")
+    private String keyStorePassword;
+
+    @Value("${ignite.trust-store-path:#{null}}")
+    private String trustStorePath;
+
+    @Value("${ignite.trust-store-password:#{null}}")
+    private String trustStorePassword;
+
+    @Bean
+    public IgniteClient igniteClientInstance() {
+        ClientConfiguration cfg = new ClientConfiguration()
+                .setAddresses(host + ":" + port);
+
+        if (keyStorePath != null && keyStorePassword != null &&
+                trustStorePath != null && trustStorePassword != null) {
+            cfg.setSslClientCertificateKeyStorePath(keyStorePath)
+                    .setSslClientCertificateKeyStorePassword(keyStorePassword)
+                    .setSslTrustCertificateKeyStorePath(trustStorePath)
+                    .setSslTrustCertificateKeyStorePassword(trustStorePassword);
+        }
+
+        IgniteClient igniteClient = null;
+
+        try {
+            igniteClient = Ignition.startClient(cfg);
+        }
+        catch (ClientException e) {
+            logger.log(Level.WARNING, "Start client failed:" + e.getMessage());
+        }
+        catch (Exception e) {
+            logger.log(Level.WARNING, "Unexpected failure:" + e.getMessage());
+        }
+
+        Assert.notNull(igniteClient, "IgniteClient is null");
+
+        return igniteClient;
+    }
+}