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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
