This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch deleteSession
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9b0869880828ff5c85386dc35f682d2d5b02c7b0
Author: lostluck <13907733+lostl...@users.noreply.github.com>
AuthorDate: Wed Mar 29 08:26:11 2023 -0700
Remove dead session code & noisy logging (#22241)
---
sdks/go/pkg/beam/core/runtime/harness/datamgr.go | 4 -
sdks/go/pkg/beam/core/runtime/harness/harness.go | 22 +--
sdks/go/pkg/beam/core/runtime/harness/logging.go | 2 -
sdks/go/pkg/beam/core/runtime/harness/session.go | 232 -----------------------
4 files changed, 9 insertions(+), 251 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/datamgr.go
b/sdks/go/pkg/beam/core/runtime/harness/datamgr.go
index 6d9b9c00452..3850ee3a366 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/datamgr.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/datamgr.go
@@ -378,8 +378,6 @@ func (c *DataChannel) read(ctx context.Context) {
return
}
- recordStreamReceive(msg)
-
// Each message may contain segments for multiple streams, so we
// must treat each segment in isolation. We maintain a local
cache
// to reduce lock contention.
@@ -533,7 +531,6 @@ type dataWriter struct {
// send requires the ch.mu lock to be held.
func (w *dataWriter) send(msg *fnpb.Elements) error {
- recordStreamSend(msg)
if err := w.ch.client.Send(msg); err != nil {
if err == io.EOF {
log.Warnf(context.TODO(), "dataWriter[%v;%v] EOF on
send; fetching real error", w.id, w.ch.id)
@@ -653,7 +650,6 @@ type timerWriter struct {
// send requires the ch.mu lock to be held.
func (w *timerWriter) send(msg *fnpb.Elements) error {
- recordStreamSend(msg)
if err := w.ch.client.Send(msg); err != nil {
if err == io.EOF {
log.Warnf(context.TODO(), "timerWriter[%v;%v] EOF on
send; fetching real error", w.id, w.ch.id)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go
b/sdks/go/pkg/beam/core/runtime/harness/harness.go
index c260a46c80e..16a172003a0 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/harness.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go
@@ -37,7 +37,6 @@ import (
fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
"github.com/apache/beam/sdks/v2/go/pkg/beam/util/diagnostics"
"github.com/apache/beam/sdks/v2/go/pkg/beam/util/grpcx"
- "github.com/golang/protobuf/proto"
"golang.org/x/sync/singleflight"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/durationpb"
@@ -91,8 +90,6 @@ func Main(ctx context.Context, loggingEndpoint,
controlEndpoint string) error {
go diagnostics.SampleForHeapProfile(ctx,
samplingFrequencySeconds, maxTimeBetweenDumpsSeconds)
}
- recordHeader()
-
// Connect to FnAPI control server. Receive and execute work.
conn, err := dial(ctx, controlEndpoint, "control", 60*time.Second)
if err != nil {
@@ -125,7 +122,8 @@ func Main(ctx context.Context, loggingEndpoint,
controlEndpoint string) error {
go func() {
defer wg.Done()
for resp := range respc {
- log.Debugf(ctx, "RESP: %v",
proto.MarshalTextString(resp))
+ // TODO(lostluck): 2023/03/29 fix debug level logging
to be flagged.
+ // log.Debugf(ctx, "RESP: %v",
proto.MarshalTextString(resp))
if err := stub.Send(resp); err != nil {
log.Errorf(ctx, "control.Send: Failed to
respond: %v", err)
@@ -177,24 +175,20 @@ func Main(ctx context.Context, loggingEndpoint,
controlEndpoint string) error {
close(respc)
wg.Wait()
if err == io.EOF {
- recordFooter()
return nil
}
return errors.Wrapf(err, "control.Recv failed")
}
// Launch a goroutine to handle the control message.
- // TODO(wcn): implement a rate limiter for 'heavy' messages?
fn := func(ctx context.Context, req *fnpb.InstructionRequest) {
- log.Debugf(ctx, "RECV: %v",
proto.MarshalTextString(req))
- recordInstructionRequest(req)
-
+ // TODO(lostluck): 2023/03/29 fix debug level logging
to be flagged.
+ // log.Debugf(ctx, "RECV: %v",
proto.MarshalTextString(req))
ctx = hooks.RunRequestHooks(ctx, req)
resp := ctrl.handleInstruction(ctx, req)
hooks.RunResponseHooks(ctx, req, resp)
- recordInstructionResponse(resp)
if resp != nil && atomic.LoadInt32(&shutdown) == 0 {
respc <- resp
}
@@ -376,9 +370,10 @@ func (c *control) handleInstruction(ctx context.Context,
req *fnpb.InstructionRe
msg := req.GetProcessBundle()
// NOTE: the harness sends a 0-length process bundle request to
sources (changed?)
-
bdID := bundleDescriptorID(msg.GetProcessBundleDescriptorId())
- log.Debugf(ctx, "PB [%v]: %v", instID, msg)
+
+ // TODO(lostluck): 2023/03/29 fix debug level logging to be
flagged.
+ // log.Debugf(ctx, "PB [%v]: %v", instID, msg)
plan, err := c.getOrCreatePlan(bdID)
// Make the plan active.
@@ -546,7 +541,8 @@ func (c *control) handleInstruction(ctx context.Context,
req *fnpb.InstructionRe
case req.GetProcessBundleSplit() != nil:
msg := req.GetProcessBundleSplit()
- log.Debugf(ctx, "PB Split: %v", msg)
+ // TODO(lostluck): 2023/03/29 fix debug level logging to be
flagged.
+ // log.Debugf(ctx, "PB Split: %v", msg)
ref := instructionID(msg.GetInstructionId())
plan, _, resp := c.getPlanOrResponse(ctx, "split", instID, ref)
diff --git a/sdks/go/pkg/beam/core/runtime/harness/logging.go
b/sdks/go/pkg/beam/core/runtime/harness/logging.go
index f9f0a65b365..dbdf66bfba4 100644
--- a/sdks/go/pkg/beam/core/runtime/harness/logging.go
+++ b/sdks/go/pkg/beam/core/runtime/harness/logging.go
@@ -183,8 +183,6 @@ func (w *remoteWriter) connect(ctx context.Context) error {
LogEntries: []*fnpb.LogEntry{msg},
}
- recordLogEntries(list)
-
if err := client.Send(list); err != nil {
if err == io.EOF {
(&log.Standard{}).Log(ctx, log.SevInfo, 0,
msg.GetMessage())
diff --git a/sdks/go/pkg/beam/core/runtime/harness/session.go
b/sdks/go/pkg/beam/core/runtime/harness/session.go
deleted file mode 100644
index a72528024d5..00000000000
--- a/sdks/go/pkg/beam/core/runtime/harness/session.go
+++ /dev/null
@@ -1,232 +0,0 @@
-// 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 harness
-
-import (
- "context"
- "fmt"
- "io"
- "sync"
-
- "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
-
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/harness/session"
- "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/hooks"
- "github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
- fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
- "github.com/golang/protobuf/proto"
-)
-
-// capture is set by the capture hook below.
-var capture io.WriteCloser
-
-var (
- sessionLock sync.Mutex
- bufPool = sync.Pool{
- New: func() any {
- return proto.NewBuffer(nil)
- },
- }
-)
-
-func isEnabled(option string) bool {
- return runtime.GlobalOptions.Get(option) == "true"
-}
-
-func recordMessage(opcode session.Kind, pb *session.Entry) error {
- if !isEnabled("session_recording") {
- return nil
- }
-
- // This is called inline with the message handling code.
- // It'd be nicer to be a bit more well-behaved and not block the main
thread
- // of execution. However, this is for recording profiles, and shouldn't
be called
- // when measuring performance, so maybe this perf hit isn't a big deal.
-
- // The format of the file is a sequence of message elements. Each
element consists of
- // three parts.
- // 1) Varint encoded length of the EntryHeader
- // 2) Encoded EntryHeader message. This is a lightweight message
designed to contain
- // enough information for a consumer to determine whether the Entry
is of interest
- // which allows optionally skipping that expensive decode.
- // 3) Encoded Entry message.
-
- body := bufPool.Get().(*proto.Buffer)
- defer bufPool.Put(body)
- if err := body.Marshal(pb); err != nil {
- return errors.Wrap(err, "unable to marshal message for session
recording")
- }
-
- eh := &session.EntryHeader{
- Kind: pb.Kind,
- Len: int64(len(body.Bytes())),
- }
-
- hdr := bufPool.Get().(*proto.Buffer)
- defer bufPool.Put(hdr)
- if err := hdr.Marshal(eh); err != nil {
- return errors.Wrap(err, "unable to marshal message header for
session recording")
- }
-
- l := bufPool.Get().(*proto.Buffer)
- defer bufPool.Put(l)
- if err := l.EncodeVarint(uint64(len(hdr.Bytes()))); err != nil {
- return errors.Wrap(err, "unable to write entry header length")
- }
-
- // Acquire the lock to write the file.
- sessionLock.Lock()
- defer sessionLock.Unlock()
-
- if _, err := capture.Write(l.Bytes()); err != nil {
- return errors.Wrap(err, "unable to write entry header length")
- }
- if _, err := capture.Write(hdr.Bytes()); err != nil {
- return errors.Wrap(err, "unable to write entry header")
- }
- if _, err := capture.Write(body.Bytes()); err != nil {
- return errors.Wrap(err, "unable to write entry body")
- }
- return nil
-}
-
-func recordInstructionRequest(req *fnpb.InstructionRequest) error {
- return recordMessage(session.Kind_INSTRUCTION_REQUEST,
- &session.Entry{
- Kind: session.Kind_INSTRUCTION_REQUEST,
- Msg: &session.Entry_InstReq{
- InstReq: req,
- },
- })
-}
-
-func recordInstructionResponse(resp *fnpb.InstructionResponse) error {
- return recordMessage(session.Kind_INSTRUCTION_RESPONSE,
- &session.Entry{
- Kind: session.Kind_INSTRUCTION_RESPONSE,
- Msg: &session.Entry_InstResp{
- InstResp: resp,
- },
- })
-}
-
-func recordStreamReceive(data *fnpb.Elements) error {
- return recordMessage(session.Kind_DATA_RECEIVED,
- &session.Entry{
- Kind: session.Kind_DATA_RECEIVED,
- Msg: &session.Entry_Elems{
- Elems: data,
- },
- })
-}
-
-func recordStreamSend(data *fnpb.Elements) error {
- return recordMessage(session.Kind_DATA_SENT,
- &session.Entry{
- Kind: session.Kind_DATA_SENT,
- Msg: &session.Entry_Elems{
- Elems: data,
- },
- })
-}
-
-func recordLogEntries(entries *fnpb.LogEntry_List) error {
- return recordMessage(session.Kind_LOG_ENTRIES,
- &session.Entry{
- Kind: session.Kind_LOG_ENTRIES,
- Msg: &session.Entry_LogEntries{
- LogEntries: entries,
- },
- })
-
-}
-
-func recordHeader() error {
- return recordMessage(session.Kind_HEADER,
- &session.Entry{
- Kind: session.Kind_HEADER,
- Msg: &session.Entry_Header{
- Header: &session.Header{
- Version: "0.0.1",
- MaxMsgLen: 4000000, // TODO(wcn): get
from DataChannelManager.
- },
- },
- })
-}
-
-// TODO(wcn): footer is designed to be the last thing recorded in the log.
However,
-// there's currently no coordination with the logging channel, so this isn't
true.
-func recordFooter() error {
- return recordMessage(session.Kind_FOOTER, &session.Entry{
- Kind: session.Kind_FOOTER,
- Msg: &session.Entry_Footer{
- Footer: &session.Footer{},
- },
- })
-}
-
-// CaptureHook writes the messaging content consumed and
-// produced by the worker, allowing the data to be used as
-// an input for the session runner. Since workers can exist
-// in a variety of environments, this allows the runner
-// to tailor the behavior best for its particular needs.
-type CaptureHook io.WriteCloser
-
-// CaptureHookFactory produces a CaptureHook from the supplied
-// options.
-type CaptureHookFactory func([]string) CaptureHook
-
-var captureHookRegistry = make(map[string]CaptureHookFactory)
-
-func init() {
- hf := func(opts []string) hooks.Hook {
- return hooks.Hook{
- Init: func(ctx context.Context) (context.Context,
error) {
- if len(opts) > 0 {
- name, opts := hooks.Decode(opts[0])
- capture =
captureHookRegistry[name](opts)
- }
- return ctx, nil
- },
- }
- }
-
- hooks.RegisterHook("session", hf)
-}
-
-// RegisterCaptureHook registers a CaptureHookFactory for the
-// supplied identifier.
-func RegisterCaptureHook(name string, c CaptureHookFactory) {
- if _, exists := captureHookRegistry[name]; exists {
- panic(fmt.Sprintf("RegisterSessionCaptureHook: %s registered
twice", name))
- }
- captureHookRegistry[name] = c
-}
-
-// EnableCaptureHook is called to request the use of a hook in a pipeline.
-// It updates the supplied pipelines to capture this request.
-func EnableCaptureHook(name string, opts []string) {
- if _, exists := captureHookRegistry[name]; !exists {
- panic(fmt.Sprintf("EnableHook: %s not registered", name))
- }
- if exists, opts := hooks.IsEnabled("session"); exists {
- n, _ := hooks.Decode(opts[0])
- if n != name {
- panic(fmt.Sprintf("EnableHook: can't enable hook %s,
hook %s already enabled", name, n))
- }
- }
-
- hooks.EnableHook("session", hooks.Encode(name, opts))
-}
