[GitHub] [beam] eantyshev commented on a diff in pull request #23461: [Playground] [Backend] Update playground cache component to increase fault tolerance

[GitBox <gi...@apache.org>]

eantyshev commented on code in PR #23461:
URL: https://github.com/apache/beam/pull/23461#discussion_r985689272


##########
playground/backend/internal/components/cache_component.go:
##########
@@ -37,73 +38,129 @@ func NewService(cache cache.Cache, db db.Database) *CacheComponent {
 
 // GetSdkCatalogFromCacheOrDatastore returns the sdk catalog from the cache
 // - If there is no sdk catalog in the cache, gets the sdk catalog from the Cloud Datastore and saves it to the cache
-func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context) ([]*entity.SDKEntity, error) {
-	sdks, err := cp.cache.GetSdkCatalog(ctx)
-	if err != nil {
-		logger.Errorf("error during getting the sdk catalog from the cache, err: %s", err.Error())
-		sdks, err = cp.db.GetSDKs(ctx)
+func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context, cacheRequestTimeout time.Duration) ([]*entity.SDKEntity, error) {

Review Comment:
   let's make `timeout` a parameter of a CacheComponent, what you think?



##########
playground/backend/README.md:
##########
@@ -94,6 +94,7 @@ default value and there is no need to set them up to launch locally:
 - `SDK_CONFIG` - is the sdk configuration file path, e.g. default example for corresponding sdk. It will be saved to cloud datastore during application startup (default value = `../sdks.yaml`)
 - `DATASTORE_EMULATOR_HOST` - is the datastore emulator address. If it is given in the environment, the application will connect to the datastore emulator.
 - `PROPERTY_PATH` - is the application properties path (default value = `.`)
+- `CACHE_REQUEST_TIMEOUT` - is the timeout to request data from cache (default value = `30 sec`)

Review Comment:
   It should not be longer than HTTP timeout, we need to check these values on startup



##########
playground/backend/internal/components/cache_component.go:
##########
@@ -37,73 +38,129 @@ func NewService(cache cache.Cache, db db.Database) *CacheComponent {
 
 // GetSdkCatalogFromCacheOrDatastore returns the sdk catalog from the cache
 // - If there is no sdk catalog in the cache, gets the sdk catalog from the Cloud Datastore and saves it to the cache
-func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context) ([]*entity.SDKEntity, error) {
-	sdks, err := cp.cache.GetSdkCatalog(ctx)
-	if err != nil {
-		logger.Errorf("error during getting the sdk catalog from the cache, err: %s", err.Error())
-		sdks, err = cp.db.GetSDKs(ctx)
+func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context, cacheRequestTimeout time.Duration) ([]*entity.SDKEntity, error) {
+	sdksCh := make(chan []*entity.SDKEntity, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		sdks, err := cp.cache.GetSdkCatalog(ctx)
 		if err != nil {
-			logger.Errorf("error during getting the sdk catalog from the cloud datastore, err: %s", err.Error())
-			return nil, err
-		}
-		if err = cp.cache.SetSdkCatalog(ctx, sdks); err != nil {
-			logger.Errorf("error during setting the sdk catalog to the cache, err: %s", err.Error())
-			return nil, err
+			errCh <- err
+		} else {
+			sdksCh <- sdks
 		}
+	}()
+	select {
+	case sdks := <-sdksCh:
+		return sdks, nil
+	case e := <-errCh:
+		logger.Errorf("error during getting the sdk catalog from the cache, err: %s", e.Error())
+		return cp.getSdks(ctx)
+	case <-time.After(cacheRequestTimeout):
+		logger.Errorf("error during getting the sdk catalog from the cache: timeout")
+		return cp.getSdks(ctx)
+	}
+}
+
+func (cp *CacheComponent) getSdks(ctx context.Context) ([]*entity.SDKEntity, error) {
+	sdks, err := cp.db.GetSDKs(ctx)
+	if err != nil {
+		logger.Errorf("error during getting the sdk catalog from the cloud datastore, err: %s", err.Error())
+		return nil, err
+	}
+	if err = cp.cache.SetSdkCatalog(ctx, sdks); err != nil {
+		logger.Errorf("error during setting the sdk catalog to the cache, err: %s", err.Error())
 	}
 	return sdks, nil
 }
 
 // GetCatalogFromCacheOrDatastore returns the example catalog from cache
 // - If there is no catalog in the cache, gets the catalog from the Cloud Datastore and saves it to the cache
-func (cp *CacheComponent) GetCatalogFromCacheOrDatastore(ctx context.Context) ([]*pb.Categories, error) {
-	catalog, err := cp.cache.GetCatalog(ctx)
-	if err != nil {
-		logger.Errorf("error during getting the catalog from the cache, err: %s", err.Error())
-		sdkCatalog, err := cp.GetSdkCatalogFromCacheOrDatastore(ctx)
+func (cp *CacheComponent) GetCatalogFromCacheOrDatastore(ctx context.Context, cacheRequestTimeout time.Duration) ([]*pb.Categories, error) {
+	catCh := make(chan []*pb.Categories, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		catalog, err := cp.cache.GetCatalog(ctx)
 		if err != nil {
-			logger.Errorf("error during getting the sdk catalog from the cache or datastore, err: %s", err.Error())
-			return nil, err
-		}
-		catalog, err = cp.db.GetCatalog(ctx, sdkCatalog)
-		if err != nil {
-			return nil, err
-		}
-		if len(catalog) == 0 {
-			logger.Warn("example catalog is empty")
-			return catalog, nil
-		}
-		if err = cp.cache.SetCatalog(ctx, catalog); err != nil {
-			logger.Errorf("SetCatalog(): cache error: %s", err.Error())
-			return nil, err
+			errCh <- err
+		} else {
+			catCh <- catalog
 		}
+	}()
+	select {
+	case cat := <-catCh:
+		return cat, nil
+	case e := <-errCh:
+		logger.Errorf("error during getting the catalog from the cache, err: %s", e.Error())
+		return cp.getCatalog(ctx, cacheRequestTimeout)

Review Comment:
   timeout is a part of a context
   https://pkg.go.dev/context#WithDeadline



##########
playground/backend/internal/components/cache_component.go:
##########
@@ -37,73 +38,129 @@ func NewService(cache cache.Cache, db db.Database) *CacheComponent {
 
 // GetSdkCatalogFromCacheOrDatastore returns the sdk catalog from the cache
 // - If there is no sdk catalog in the cache, gets the sdk catalog from the Cloud Datastore and saves it to the cache
-func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context) ([]*entity.SDKEntity, error) {
-	sdks, err := cp.cache.GetSdkCatalog(ctx)
-	if err != nil {
-		logger.Errorf("error during getting the sdk catalog from the cache, err: %s", err.Error())
-		sdks, err = cp.db.GetSDKs(ctx)
+func (cp *CacheComponent) GetSdkCatalogFromCacheOrDatastore(ctx context.Context, cacheRequestTimeout time.Duration) ([]*entity.SDKEntity, error) {
+	sdksCh := make(chan []*entity.SDKEntity, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		sdks, err := cp.cache.GetSdkCatalog(ctx)

Review Comment:
   let's just do cache request with a reduced timeout:
   ```
   ctx = context.WithTimeout(ctx, CACHE_TO)
   ```
   
   no need for goroutines here



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org