Merge branch 'main' into simple_java_client

Author: seeflood

README.md

@@ -9,14 +9,16 @@
 
 Layotto is an application runtime developed using Golang, which provides various distributed capabilities for applications, such as state management, configuration management, and event pub/sub capabilities to simplify application development.
 
-Layotto uses the open source [MOSN](https://github.com/mosn/mosn) as the base, in addition to providing distributed capabilities, it also provides Service Mesh's ability to control traffic.
+Layotto is built on the open source data plane [MOSN](https://github.com/mosn/mosn) .In addition to providing distributed building blocks, Layotto can also serve as the data plane of Service Mesh and has the ability to control traffic.
 
 ## Motivation
 
-Layotto aims to combine Runtime with Service Mesh into one sidecar. No matter which product you are using as the Service Mesh data plane (e.g. MOSN,Envoy or any other product), you can always attach Layotto to it and add Multi-Runtime capabilities without adding new sidecars. 
+Layotto aims to combine [Multi-Runtime](https://www.infoq.com/articles/multi-runtime-microservice-architecture/) with Service Mesh into one sidecar. No matter which product you are using as the Service Mesh data plane (e.g. MOSN,Envoy or any other product), you can always attach Layotto to it and add Multi-Runtime capabilities without adding new sidecars. 
 
 For example, by adding Runtime capabilities to MOSN, a Layotto process can both [serve as the data plane of istio](https://mosn.io/layotto/#/en/start/istio/start.md) and provide various Runtime APIs (such as Configuration API, Pub/Sub API, etc.)
 
+In addition, we were surprised to find that a sidecar can do much more than that. We are trying to make Layotto even the runtime container of FaaS (Function as a service) and [reloadable sdk](https://github.com/mosn/layotto/issues/166) with the magic power of [WebAssembly](https://en.wikipedia.org/wiki/WebAssembly) .
+
 ## Features
 
 - Service Communication
@@ -32,7 +34,7 @@ For example, by adding Runtime capabilities to MOSN, a Layotto process can both
 
 As shown in the architecture diagram below, Layotto uses the open source MOSN as the base to provide network layer management capabilities while providing distributed capabilities. The business logic can directly interact with Layotto through a lightweight SDK without paying attention to the specific back-end infrastructure.
 
-Layotto provides sdk in various languages. The sdk interacts with Layotto through grpc. Application developers only need to specify their own infrastructure type through the configuration file [configure file](./configs/runtime_config.json) provided by Layotto. No coding changes are required, which greatly improves the portability of the program.
+Layotto provides sdks in various languages. The sdk interacts with Layotto through grpc. Application developers only need to specify their own infrastructure type through the configuration file [configure file](./configs/runtime_config.json) provided by Layotto. No coding changes are required, which greatly improves the portability of the program.
 
 ![Architecture](https://raw.githubusercontent.com/mosn/layotto/main/docs/img/runtime-architecture.png)
 

cmd/layotto/main.go

@@ -88,7 +88,8 @@ import (
 
 	// Sequencer
 	sequencer_etcd "mosn.io/layotto/components/sequencer/etcd"
-
+	sequencer_redis "mosn.io/layotto/components/sequencer/redis"
+	sequencer_zookeeper "mosn.io/layotto/components/sequencer/zookeeper"
 	// Actuator
 	_ "mosn.io/layotto/pkg/actuator"
 	"mosn.io/layotto/pkg/actuator/health"
@@ -263,6 +264,12 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_sequencer.NewFactory("etcd", func() sequencer.Store {
 				return sequencer_etcd.NewEtcdSequencer(log.DefaultLogger)
 			}),
+			runtime_sequencer.NewFactory("redis", func() sequencer.Store {
+				return sequencer_redis.NewStandaloneRedisSequencer(log.DefaultLogger)
+			}),
+			runtime_sequencer.NewFactory("zookeeper", func() sequencer.Store {
+				return sequencer_zookeeper.NewZookeeperSequencer(log.DefaultLogger)
+			}),
 		))
 	// 4. check if unhealthy
 	if err != nil {

components/lock/etcd/etcd_lock.go

@@ -2,38 +2,17 @@ package etcd
 
 import (
 	"context"
-	"crypto/tls"
-	"crypto/x509"
-	"errors"
 	"fmt"
-	"io/ioutil"
-	"strconv"
-	"strings"
-	"time"
-
 	"go.etcd.io/etcd/client/v3"
+	"mosn.io/layotto/components/pkg/utils"
 
 	"mosn.io/layotto/components/lock"
 	"mosn.io/pkg/log"
 )
 
-const (
-	defaultDialTimeout = 5
-	defaultKeyPrefix   = "/layotto/"
-
-	prefixKey         = "keyPrefixPath"
-	usernameKey       = "username"
-	passwordKey       = "password"
-	dialTimeoutKey    = "dialTimeout"
-	endpointsKey      = "endpoints"
-	tlsCertPathKey    = "tlsCert"
-	tlsCertKeyPathKey = "tlsCertKey"
-	tlsCaPathKey      = "tlsCa"
-)
-
 type EtcdLock struct {
 	client   *clientv3.Client
-	metadata metadata
+	metadata utils.EtcdMetadata
 
 	features []lock.Feature
 	logger   log.ErrorLogger
@@ -54,13 +33,13 @@ func NewEtcdLock(logger log.ErrorLogger) *EtcdLock {
 
 func (e *EtcdLock) Init(metadata lock.Metadata) error {
 	// 1. parse config
-	m, err := parseEtcdMetadata(metadata)
+	m, err := utils.ParseEtcdMetadata(metadata.Properties)
 	if err != nil {
 		return err
 	}
 	e.metadata = m
 	// 2. construct client
-	if e.client, err = e.newClient(m); err != nil {
+	if e.client, err = utils.NewEtcdClient(m); err != nil {
 		return err
 	}
 
@@ -134,130 +113,12 @@ func (e *EtcdLock) Close() error {
 	return e.client.Close()
 }
 
-func (e *EtcdLock) newClient(meta metadata) (*clientv3.Client, error) {
-
-	config := clientv3.Config{
-		Endpoints:   meta.endpoints,
-		DialTimeout: time.Second * time.Duration(meta.dialTimeout),
-		Username:    meta.username,
-		Password:    meta.password,
-	}
-
-	if meta.tlsCa != "" || meta.tlsCert != "" || meta.tlsCertKey != "" {
-		//enable tls
-		cert, err := tls.LoadX509KeyPair(meta.tlsCert, meta.tlsCertKey)
-		if err != nil {
-			return nil, fmt.Errorf("error reading tls certificate, cert: %s, certKey: %s, err: %s", meta.tlsCert, meta.tlsCertKey, err)
-		}
-
-		caData, err := ioutil.ReadFile(meta.tlsCa)
-		if err != nil {
-			return nil, fmt.Errorf("error reading tls ca %s, err: %s", meta.tlsCa, err)
-		}
-
-		pool := x509.NewCertPool()
-		pool.AppendCertsFromPEM(caData)
-
-		tlsConfig := &tls.Config{
-			Certificates: []tls.Certificate{cert},
-			RootCAs:      pool,
-		}
-		config.TLS = tlsConfig
-	}
-
-	if client, err := clientv3.New(config); err != nil {
-		return nil, err
-	} else {
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second*time.Duration(meta.dialTimeout))
-		defer cancel()
-		//ping
-		_, err = client.Get(ctx, "ping")
-		if err != nil {
-			return nil, fmt.Errorf("etcd lock error: connect to etcd timeoout %s", meta.endpoints)
-		}
-
-		return client, nil
-	}
-}
-
 func (e *EtcdLock) getKey(resourceId string) string {
-	return fmt.Sprintf("%s%s", e.metadata.keyPrefix, resourceId)
+	return fmt.Sprintf("%s%s", e.metadata.KeyPrefix, resourceId)
 }
 
 func newInternalErrorUnlockResponse() *lock.UnlockResponse {
 	return &lock.UnlockResponse{
 		Status: lock.INTERNAL_ERROR,
 	}
 }
-
-func parseEtcdMetadata(meta lock.Metadata) (metadata, error) {
-	m := metadata{}
-	var err error
-
-	if val, ok := meta.Properties[endpointsKey]; ok && val != "" {
-		m.endpoints = strings.Split(val, ";")
-	} else {
-		return m, errors.New("etcd lock error: missing endpoints address")
-	}
-
-	if val, ok := meta.Properties[dialTimeoutKey]; ok && val != "" {
-		if m.dialTimeout, err = strconv.Atoi(val); err != nil {
-			return m, fmt.Errorf("etcd lock error: ncorrect dialTimeout value %s", val)
-		}
-	} else {
-		m.dialTimeout = defaultDialTimeout
-	}
-
-	if val, ok := meta.Properties[prefixKey]; ok && val != "" {
-		m.keyPrefix = addPathSeparator(val)
-	} else {
-		m.keyPrefix = defaultKeyPrefix
-	}
-
-	if val, ok := meta.Properties[usernameKey]; ok && val != "" {
-		m.username = val
-	}
-
-	if val, ok := meta.Properties[passwordKey]; ok && val != "" {
-		m.password = val
-	}
-
-	if val, ok := meta.Properties[tlsCaPathKey]; ok && val != "" {
-		m.tlsCa = val
-	}
-
-	if val, ok := meta.Properties[tlsCertPathKey]; ok && val != "" {
-		m.tlsCert = val
-	}
-
-	if val, ok := meta.Properties[tlsCertKeyPathKey]; ok && val != "" {
-		m.tlsCertKey = val
-	}
-
-	return m, nil
-}
-
-func addPathSeparator(p string) string {
-	if p == "" {
-		return "/"
-	}
-	if p[0] != '/' {
-		p = "/" + p
-	}
-	if p[len(p)-1] != '/' {
-		p = p + "/"
-	}
-	return p
-}
-
-type metadata struct {
-	keyPrefix   string
-	dialTimeout int
-	endpoints   []string
-	username    string
-	password    string
-
-	tlsCa      string
-	tlsCert    string
-	tlsCertKey string
-}

components/lock/redis/standalone_redis_lock.go

@@ -2,34 +2,18 @@ package redis
 
 import (
 	"context"
-	"crypto/tls"
-	"errors"
 	"fmt"
 	"github.com/go-redis/redis/v8"
 	"mosn.io/layotto/components/lock"
+	"mosn.io/layotto/components/pkg/utils"
 	"mosn.io/pkg/log"
-	"strconv"
 	"time"
 )
 
-const (
-	host                   = "redisHost"
-	password               = "redisPassword"
-	enableTLS              = "enableTLS"
-	maxRetries             = "maxRetries"
-	maxRetryBackoff        = "maxRetryBackoff"
-	defaultBase            = 10
-	defaultBitSize         = 0
-	defaultDB              = 0
-	defaultMaxRetries      = 3
-	defaultMaxRetryBackoff = time.Second * 2
-	defaultEnableTLS       = false
-)
-
 // Standalone Redis lock store.Any fail-over related features are not supported,such as Sentinel and Redis Cluster.
 type StandaloneRedisLock struct {
 	client   *redis.Client
-	metadata metadata
+	metadata utils.RedisMetadata
 	replicas int
 
 	features []lock.Feature
@@ -51,37 +35,21 @@ func NewStandaloneRedisLock(logger log.ErrorLogger) *StandaloneRedisLock {
 
 func (p *StandaloneRedisLock) Init(metadata lock.Metadata) error {
 	// 1. parse config
-	m, err := parseRedisMetadata(metadata)
+	m, err := utils.ParseRedisMetadata(metadata.Properties)
 	if err != nil {
 		return err
 	}
 	p.metadata = m
 	// 2. construct client
-	p.client = p.newClient(m)
+	p.client = utils.NewRedisClient(m)
 	p.ctx, p.cancel = context.WithCancel(context.Background())
 	// 3. connect to redis
 	if _, err = p.client.Ping(p.ctx).Result(); err != nil {
-		return fmt.Errorf("[standaloneRedisLock]: error connecting to redis at %s: %s", m.host, err)
+		return fmt.Errorf("[standaloneRedisLock]: error connecting to redis at %s: %s", m.Host, err)
 	}
 	return err
 }
 
-func (p *StandaloneRedisLock) newClient(m metadata) *redis.Client {
-	opts := &redis.Options{
-		Addr:            m.host,
-		Password:        m.password,
-		DB:              defaultDB,
-		MaxRetries:      m.maxRetries,
-		MaxRetryBackoff: m.maxRetryBackoff,
-	}
-	if m.enableTLS {
-		opts.TLSConfig = &tls.Config{
-			InsecureSkipVerify: m.enableTLS,
-		}
-	}
-	return redis.NewClient(opts)
-}
-
 func (p *StandaloneRedisLock) Features() []lock.Feature {
 	return p.features
 }
@@ -145,54 +113,3 @@ func (p *StandaloneRedisLock) Close() error {
 
 	return p.client.Close()
 }
-
-func parseRedisMetadata(meta lock.Metadata) (metadata, error) {
-	m := metadata{}
-
-	if val, ok := meta.Properties[host]; ok && val != "" {
-		m.host = val
-	} else {
-		return m, errors.New("redis store error: missing host address")
-	}
-
-	if val, ok := meta.Properties[password]; ok && val != "" {
-		m.password = val
-	}
-
-	m.enableTLS = defaultEnableTLS
-	if val, ok := meta.Properties[enableTLS]; ok && val != "" {
-		tls, err := strconv.ParseBool(val)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse enableTLS field: %s", err)
-		}
-		m.enableTLS = tls
-	}
-
-	m.maxRetries = defaultMaxRetries
-	if val, ok := meta.Properties[maxRetries]; ok && val != "" {
-		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse maxRetries field: %s", err)
-		}
-		m.maxRetries = int(parsedVal)
-	}
-
-	m.maxRetryBackoff = defaultMaxRetryBackoff
-	if val, ok := meta.Properties[maxRetryBackoff]; ok && val != "" {
-		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse maxRetryBackoff field: %s", err)
-		}
-		m.maxRetryBackoff = time.Duration(parsedVal)
-	}
-
-	return m, nil
-}
-
-type metadata struct {
-	host            string
-	password        string
-	maxRetries      int
-	maxRetryBackoff time.Duration
-	enableTLS       bool
-}