You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by pa...@apache.org on 2021/11/23 03:23:33 UTC
[beam] branch master updated: Merge pull request #15996 from [BEAM-13156][Playground] Implement streaming output for the run code logs
This is an automated email from the ASF dual-hosted git repository.
pabloem pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b78de49 Merge pull request #15996 from [BEAM-13156][Playground] Implement streaming output for the run code logs
b78de49 is described below
commit b78de493a4a5f56cf8339ab2f174cc25cc57beec
Author: Aydar Zainutdinov <ay...@akvelon.com>
AuthorDate: Tue Nov 23 06:22:27 2021 +0300
Merge pull request #15996 from [BEAM-13156][Playground] Implement streaming output for the run code logs
* [Playground][BEAM-12941][Bugfix] Fix workflows for playground applications (#83)
* Update workflows for playground
* Attempt to fix tests
* Remove continue on error to catch errors
* Fix linter problem for backend dockerfile
* Update folder to run backend go linter
* Moved flutter test to execution via gradle tasks
* Revert "[Playground][BEAM-12941][Bugfix] Fix workflows for playground applications (#83)" (#88)
This reverts commit b73f5f70ac4184b56a0a03922731c5f2f69b9566.
* [BEAM-13156][Playground]
implementation of streaming output for run code output
Co-authored-by: Sergey Kalinin <91...@users.noreply.github.com>
Co-authored-by: Ilya <il...@akvelon.com>
---
playground/backend/cmd/server/controller.go | 17 ++-
playground/backend/cmd/server/controller_test.go | 16 +++
playground/backend/internal/cache/cache.go | 3 +
.../internal/code_processing/code_processing.go | 87 ++++++++++-----
.../code_processing/code_processing_test.go | 79 +++++++++++++
.../internal/streaming/run_output_writer.go | 60 ++++++++++
.../internal/streaming/run_output_writer_test.go | 123 +++++++++++++++++++++
7 files changed, 355 insertions(+), 30 deletions(-)
diff --git a/playground/backend/cmd/server/controller.go b/playground/backend/cmd/server/controller.go
index f42266c..ec24c94 100644
--- a/playground/backend/cmd/server/controller.go
+++ b/playground/backend/cmd/server/controller.go
@@ -74,6 +74,9 @@ func (controller *playgroundController) RunCode(ctx context.Context, info *pb.Ru
if err = utils.SetToCache(ctx, controller.cacheService, pipelineId, cache.Status, pb.Status_STATUS_VALIDATING); err != nil {
return nil, errors.InternalError("Run code()", "Error during set value to cache: "+err.Error())
}
+ if err = utils.SetToCache(ctx, controller.cacheService, pipelineId, cache.RunOutputIndex, 0); err != nil {
+ return nil, errors.InternalError("Run code()", "Error during set value to cache: "+err.Error())
+ }
if err = controller.cacheService.SetExpTime(ctx, pipelineId, cacheExpirationTime); err != nil {
logger.Errorf("%s: RunCode(): cache.SetExpTime(): %s\n", pipelineId, err.Error())
return nil, errors.InternalError("Run code()", "Error during set expiration to cache: "+err.Error())
@@ -107,11 +110,23 @@ func (controller *playgroundController) GetRunOutput(ctx context.Context, info *
logger.Errorf("%s: GetRunOutput(): pipelineId has incorrect value and couldn't be parsed as uuid value: %s", info.PipelineUuid, err.Error())
return nil, errors.InvalidArgumentError("GetRunOutput", "pipelineId has incorrect value and couldn't be parsed as uuid value: "+info.PipelineUuid)
}
+ lastIndex, err := code_processing.GetRunOutputLastIndex(ctx, controller.cacheService, pipelineId, "GetRunOutput")
+ if err != nil {
+ return nil, err
+ }
runOutput, err := code_processing.GetProcessingOutput(ctx, controller.cacheService, pipelineId, cache.RunOutput, "GetRunOutput")
if err != nil {
return nil, err
}
- return &pb.GetRunOutputResponse{Output: runOutput}, nil
+ newRunOutput := ""
+ if len(runOutput) > lastIndex {
+ newRunOutput = runOutput[lastIndex:]
+ utils.SetToCache(ctx, controller.cacheService, pipelineId, cache.RunOutputIndex, lastIndex+len(newRunOutput))
+ }
+
+ pipelineResult := pb.GetRunOutputResponse{Output: newRunOutput}
+
+ return &pipelineResult, nil
}
// GetRunError is returning error output of execution for specific pipeline by PipelineUuid
diff --git a/playground/backend/cmd/server/controller_test.go b/playground/backend/cmd/server/controller_test.go
index 25d17a2..0caa292 100644
--- a/playground/backend/cmd/server/controller_test.go
+++ b/playground/backend/cmd/server/controller_test.go
@@ -418,6 +418,7 @@ func TestPlaygroundController_GetRunOutput(t *testing.T) {
// As a result want to receive response with an expected run output.
name: "run output exist",
prepare: func() {
+ _ = cacheService.SetValue(ctx, pipelineId, cache.RunOutputIndex, 0)
_ = cacheService.SetValue(ctx, pipelineId, cache.RunOutput, runOutput)
},
args: args{
@@ -427,6 +428,21 @@ func TestPlaygroundController_GetRunOutput(t *testing.T) {
want: &pb.GetRunOutputResponse{Output: runOutput},
wantErr: false,
},
+ {
+ // Test case with calling RunOutput method with pipelineId which contain run output and index of run output is 1.
+ // As a result want to receive response with correct output (output[1:]).
+ name: "get the second part",
+ prepare: func() {
+ _ = cacheService.SetValue(ctx, pipelineId, cache.RunOutputIndex, 1)
+ _ = cacheService.SetValue(ctx, pipelineId, cache.RunOutput, runOutput)
+ },
+ args: args{
+ ctx: ctx,
+ info: &pb.GetRunOutputRequest{PipelineUuid: pipelineId.String()},
+ },
+ want: &pb.GetRunOutputResponse{Output: runOutput[1:]},
+ wantErr: false,
+ },
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
diff --git a/playground/backend/internal/cache/cache.go b/playground/backend/internal/cache/cache.go
index 21dc090..84549c0 100644
--- a/playground/backend/internal/cache/cache.go
+++ b/playground/backend/internal/cache/cache.go
@@ -40,6 +40,9 @@ const (
// Canceled is used to keep the canceled status
Canceled SubKey = "CANCELED"
+
+ // RunOutputIndex is the index of the start of the run step's output
+ RunOutputIndex SubKey = "RUN_OUTPUT_INDEX"
)
// Cache is used to store states and outputs for Apache Beam pipelines that running in Playground
diff --git a/playground/backend/internal/code_processing/code_processing.go b/playground/backend/internal/code_processing/code_processing.go
index d689802..0763cfe 100644
--- a/playground/backend/internal/code_processing/code_processing.go
+++ b/playground/backend/internal/code_processing/code_processing.go
@@ -24,6 +24,8 @@ import (
"beam.apache.org/playground/backend/internal/fs_tool"
"beam.apache.org/playground/backend/internal/logger"
"beam.apache.org/playground/backend/internal/setup_tools/run_builder"
+ "beam.apache.org/playground/backend/internal/streaming"
+ "bytes"
"context"
"fmt"
"github.com/google/uuid"
@@ -56,32 +58,42 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
go cancelCheck(ctxWithTimeout, pipelineId, cancelChannel, cacheService)
// build executor for validate and compile steps
- exec := compileBuilder.Build()
+ executor := compileBuilder.Build()
// validate
logger.Infof("%s: Validate() ...\n", pipelineId)
- validateFunc := exec.Validate()
+ validateFunc := executor.Validate()
go validateFunc(successChannel, errorChannel)
- if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, nil, errorChannel, pb.Status_STATUS_VALIDATION_ERROR, pb.Status_STATUS_PREPARING); err != nil {
+ if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, nil, nil, errorChannel, pb.Status_STATUS_VALIDATION_ERROR, pb.Status_STATUS_PREPARING); err != nil {
return
}
// prepare
logger.Infof("%s: Prepare() ...\n", pipelineId)
- prepareFunc := exec.Prepare()
+ prepareFunc := executor.Prepare()
go prepareFunc(successChannel, errorChannel)
- if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, nil, errorChannel, pb.Status_STATUS_PREPARATION_ERROR, pb.Status_STATUS_COMPILING); err != nil {
+ if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, nil, nil, errorChannel, pb.Status_STATUS_PREPARATION_ERROR, pb.Status_STATUS_COMPILING); err != nil {
return
}
// compile
logger.Infof("%s: Compile() ...\n", pipelineId)
- compileCmd := exec.Compile(ctxWithTimeout)
- go processCmd(compileCmd, successChannel, errorChannel, dataChannel)
+ compileCmd := executor.Compile(ctxWithTimeout)
+ go func(cmd *exec.Cmd, successChannel chan bool, errChannel chan error, dataChannel chan interface{}) {
+ // TODO separate stderr from stdout [BEAM-13208]
+ data, err := cmd.CombinedOutput()
+ dataChannel <- data
+ if err != nil {
+ errChannel <- err
+ successChannel <- false
+ } else {
+ successChannel <- true
+ }
+ }(compileCmd, successChannel, errorChannel, dataChannel)
- if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, dataChannel, errorChannel, pb.Status_STATUS_COMPILE_ERROR, pb.Status_STATUS_EXECUTING); err != nil {
+ if err := processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, dataChannel, nil, errorChannel, pb.Status_STATUS_COMPILE_ERROR, pb.Status_STATUS_EXECUTING); err != nil {
return
}
@@ -93,14 +105,27 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
}
// build executor for run step
- exec = runBuilder.Build()
+ executor = runBuilder.Build()
// run
logger.Infof("%s: Run() ...\n", pipelineId)
- runCmd := exec.Run(ctxWithTimeout)
- go processCmd(runCmd, successChannel, errorChannel, dataChannel)
+ runCmd := executor.Run(ctxWithTimeout)
+
+ var runError bytes.Buffer
+ runOutput := &streaming.RunOutputWriter{Ctx: ctxWithTimeout, CacheService: cacheService, PipelineId: pipelineId}
+ runCmd.Stderr = &runError
+ runCmd.Stdout = runOutput
+ go func(cmd *exec.Cmd, successChannel chan bool, errChannel chan error, dataChannel chan interface{}) {
+ err := cmd.Run()
+ if err != nil {
+ errChannel <- err
+ successChannel <- false
+ } else {
+ successChannel <- true
+ }
+ }(runCmd, successChannel, errorChannel, dataChannel)
- processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, dataChannel, errorChannel, pb.Status_STATUS_RUN_ERROR, pb.Status_STATUS_FINISHED)
+ processStep(ctxWithTimeout, pipelineId, cacheService, cancelChannel, successChannel, nil, &runError, errorChannel, pb.Status_STATUS_RUN_ERROR, pb.Status_STATUS_FINISHED)
}
// GetProcessingOutput gets processing output value from cache by key and subKey.
@@ -138,25 +163,27 @@ func GetProcessingStatus(ctx context.Context, cacheService cache.Cache, key uuid
return statusValue, nil
}
-// processCmd is used to run command. The output of the command running is set to dataCh.
-// In case error occurs - set error output to erCh and put false to doneCh.
-// In case command finishes without errors - put true to doneCh.
-func processCmd(cmd *exec.Cmd, doneCh chan bool, errCh chan error, dataCh chan interface{}) {
- // TODO separate stderr from stdout
- data, err := cmd.CombinedOutput()
- dataCh <- data
+// GetRunOutputLastIndex gets run output's last index from cache by key.
+// In case key doesn't exist in cache - returns an errors.NotFoundError.
+// In case value from cache by key and subKey couldn't be converted to int - returns an errors.InternalError.
+func GetRunOutputLastIndex(ctx context.Context, cacheService cache.Cache, key uuid.UUID, errorTitle string) (int, error) {
+ value, err := cacheService.GetValue(ctx, key, cache.RunOutputIndex)
if err != nil {
- errCh <- err
- doneCh <- false
- } else {
- doneCh <- true
+ logger.Errorf("%s: GetStringValueFromCache(): cache.GetValue: error: %s", key, err.Error())
+ return 0, errors.NotFoundError(errorTitle, fmt.Sprintf("Error during getting cache by key: %s, subKey: %s", key.String(), string(cache.RunOutputIndex)))
}
+ intValue, converted := value.(int)
+ if !converted {
+ logger.Errorf("%s: couldn't convert value to string: %s", key, value)
+ return 0, errors.InternalError(errorTitle, fmt.Sprintf("Value from cache couldn't be converted to int: %s", value))
+ }
+ return intValue, nil
}
// processStep processes each executor's step with cancel and timeout checks.
-// If finishes by canceling, timeout or error - returns false.
-// If finishes successfully returns true.
-func processStep(ctx context.Context, pipelineId uuid.UUID, cacheService cache.Cache, cancelChannel, successChannel chan bool, dataChannel chan interface{}, errorChannel chan error, errorCaseStatus, successCaseStatus pb.Status) error {
+// If finishes by canceling, timeout or error - returns error.
+// If finishes successfully returns nil.
+func processStep(ctx context.Context, pipelineId uuid.UUID, cacheService cache.Cache, cancelChannel, successChannel chan bool, dataChannel chan interface{}, errorDataBuffer *bytes.Buffer, errorChannel chan error, errorCaseStatus, successCaseStatus pb.Status) error {
select {
case <-ctx.Done():
finishByTimeout(ctx, pipelineId, cacheService)
@@ -172,7 +199,11 @@ func processStep(ctx context.Context, pipelineId uuid.UUID, cacheService cache.C
}
if !ok {
err := <-errorChannel
- processError(ctx, err, data, pipelineId, cacheService, errorCaseStatus)
+ var errorData = data
+ if errorDataBuffer != nil {
+ errorData = errorDataBuffer.Bytes()
+ }
+ processError(ctx, err, errorData, pipelineId, cacheService, errorCaseStatus)
return fmt.Errorf("%s: code processing finishes with error: %s", pipelineId, err.Error())
}
processSuccess(ctx, data, pipelineId, cacheService, successCaseStatus)
@@ -270,8 +301,6 @@ func processSuccess(ctx context.Context, output []byte, pipelineId uuid.UUID, ca
case pb.Status_STATUS_FINISHED:
logger.Infof("%s: Run() finish\n", pipelineId)
- cacheService.SetValue(ctx, pipelineId, cache.RunOutput, string(output))
-
cacheService.SetValue(ctx, pipelineId, cache.Status, pb.Status_STATUS_FINISHED)
}
}
diff --git a/playground/backend/internal/code_processing/code_processing_test.go b/playground/backend/internal/code_processing/code_processing_test.go
index 9cb73c9..e79d249 100644
--- a/playground/backend/internal/code_processing/code_processing_test.go
+++ b/playground/backend/internal/code_processing/code_processing_test.go
@@ -445,3 +445,82 @@ func TestGetProcessingStatus(t *testing.T) {
})
}
}
+
+func TestGetRunOutputLastIndex(t *testing.T) {
+ defer goleak.VerifyNone(t, opt)
+ pipelineId := uuid.New()
+ incorrectConvertPipelineId := uuid.New()
+ err := cacheService.SetValue(context.Background(), pipelineId, cache.RunOutputIndex, 2)
+ if err != nil {
+ panic(err)
+ }
+ err = cacheService.SetValue(context.Background(), incorrectConvertPipelineId, cache.RunOutputIndex, "MOCK_LAST_INDEX")
+ if err != nil {
+ panic(err)
+ }
+
+ type args struct {
+ ctx context.Context
+ cacheService cache.Cache
+ key uuid.UUID
+ errorTitle string
+ }
+ tests := []struct {
+ name string
+ args args
+ want int
+ wantErr bool
+ }{
+ {
+ // Test case with calling GetRunOutputLastIndex with pipelineId which doesn't contain status.
+ // As a result, want to receive an error.
+ name: "get last index with incorrect pipelineId",
+ args: args{
+ ctx: context.Background(),
+ cacheService: cacheService,
+ key: uuid.New(),
+ errorTitle: "",
+ },
+ want: 0,
+ wantErr: true,
+ },
+ {
+ // Test case with calling GetRunOutputLastIndex with pipelineId which contains incorrect status value in cache.
+ // As a result, want to receive an error.
+ name: "get last index with incorrect cache value",
+ args: args{
+ ctx: context.Background(),
+ cacheService: cacheService,
+ key: incorrectConvertPipelineId,
+ errorTitle: "",
+ },
+ want: 0,
+ wantErr: true,
+ },
+ {
+ // Test case with calling GetRunOutputLastIndex with pipelineId which contains status.
+ // As a result, want to receive an expected last index.
+ name: "get last index with correct pipelineId",
+ args: args{
+ ctx: context.Background(),
+ cacheService: cacheService,
+ key: pipelineId,
+ errorTitle: "",
+ },
+ want: 2,
+ wantErr: false,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := GetRunOutputLastIndex(tt.args.ctx, tt.args.cacheService, tt.args.key, tt.args.errorTitle)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("GetRunOutputLastIndex() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if got != tt.want {
+ t.Errorf("GetRunOutputLastIndex() got = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
diff --git a/playground/backend/internal/streaming/run_output_writer.go b/playground/backend/internal/streaming/run_output_writer.go
new file mode 100644
index 0000000..94bd0a1
--- /dev/null
+++ b/playground/backend/internal/streaming/run_output_writer.go
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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 streaming
+
+import (
+ "beam.apache.org/playground/backend/internal/cache"
+ "context"
+ "fmt"
+ "github.com/google/uuid"
+)
+
+// RunOutputWriter is used to write the run step's output to cache as a stream.
+type RunOutputWriter struct {
+ Ctx context.Context
+ CacheService cache.Cache
+ PipelineId uuid.UUID
+}
+
+// Write writes len(p) bytes from p to cache with cache.RunOutput subKey.
+// In case some error occurs - returns (0, error).
+// In case finished with no error - returns (len(p), nil).
+//
+// As a result new bytes will be added to cache with old run output value.
+// Example:
+// p = []byte(" with new run output")
+// before Write(p): {pipelineId}:cache.RunOutput = "old run output"
+// after Write(p): {pipelineId}:cache.RunOutput = "old run output with new run output"
+func (row *RunOutputWriter) Write(p []byte) (int, error) {
+ if len(p) == 0 {
+ return 0, nil
+ }
+
+ prevOutput, err := row.CacheService.GetValue(row.Ctx, row.PipelineId, cache.RunOutput)
+ if err != nil {
+ return 0, err
+ }
+
+ // concat prevValue and new value
+ str := fmt.Sprintf("%s%s", prevOutput.(string), string(p))
+
+ // set new cache value
+ err = row.CacheService.SetValue(row.Ctx, row.PipelineId, cache.RunOutput, str)
+ if err != nil {
+ return 0, err
+ }
+ return len(p), nil
+}
diff --git a/playground/backend/internal/streaming/run_output_writer_test.go b/playground/backend/internal/streaming/run_output_writer_test.go
new file mode 100644
index 0000000..f946fbd
--- /dev/null
+++ b/playground/backend/internal/streaming/run_output_writer_test.go
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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 streaming
+
+import (
+ "beam.apache.org/playground/backend/internal/cache"
+ "beam.apache.org/playground/backend/internal/cache/local"
+ "context"
+ "github.com/google/uuid"
+ "testing"
+)
+
+func TestRunOutputWriter_Write(t *testing.T) {
+ pipelineId := uuid.New()
+ cacheService := local.New(context.Background())
+ err := cacheService.SetValue(context.Background(), pipelineId, cache.RunOutput, "MOCK_TEST_OUTPUT")
+ if err != nil {
+ panic(err)
+ }
+
+ type fields struct {
+ Ctx context.Context
+ CacheService cache.Cache
+ PipelineId uuid.UUID
+ }
+ type args struct {
+ p []byte
+ }
+ tests := []struct {
+ name string
+ fields fields
+ args args
+ check func() bool
+ want int
+ wantErr bool
+ }{
+ {
+ // Test case with calling Write method with empty string parameter.
+ // As a result, want to receive 0.
+ name: "empty string",
+ fields: fields{
+ Ctx: context.Background(),
+ CacheService: cacheService,
+ PipelineId: uuid.New(),
+ },
+ args: args{[]byte("")},
+ check: func() bool {
+ return true
+ },
+ want: 0,
+ wantErr: false,
+ },
+ {
+ // Test case with calling Write method with pipelineId which doesn't contain output yet.
+ // As a result, want to receive an error.
+ name: "output doesn't exist",
+ fields: fields{
+ Ctx: context.Background(),
+ CacheService: cacheService,
+ PipelineId: uuid.New(),
+ },
+ args: args{[]byte(" NEW_MOCK_OUTPUT")},
+ check: func() bool {
+ return true
+ },
+ want: 0,
+ wantErr: true,
+ },
+ {
+ // Test case with calling Write method with pipelineId which already contains output.
+ // As a result, want to receive len of sent string.
+ name: "output exists",
+ fields: fields{
+ Ctx: context.Background(),
+ CacheService: cacheService,
+ PipelineId: pipelineId,
+ },
+ args: args{[]byte(" NEW_MOCK_OUTPUT")},
+ check: func() bool {
+ value, err := cacheService.GetValue(context.Background(), pipelineId, cache.RunOutput)
+ if err != nil {
+ return false
+ }
+ return value == "MOCK_TEST_OUTPUT NEW_MOCK_OUTPUT"
+ },
+ want: 16,
+ wantErr: false,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ row := &RunOutputWriter{
+ Ctx: tt.fields.Ctx,
+ CacheService: tt.fields.CacheService,
+ PipelineId: tt.fields.PipelineId,
+ }
+ got, err := row.Write(tt.args.p)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("Write() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if got != tt.want {
+ t.Errorf("Write() got = %v, want %v", got, tt.want)
+ }
+ if !tt.check() {
+ t.Errorf("Write() writes unexpectable data to cache")
+ }
+ })
+ }
+}