ctx, done, err := p.TrackDiskHealth(ctx, storageMetricListVols, "/")
	if err != nil {
		return nil, err
	}
	defer done(&err)

	return p.storage.ListVols(ctx)
}

func (p *xlStorageDiskIDCheck) StatVol(ctx context.Context, volume string) (vol VolInfo, err error) {
	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricStatVol, volume)
	if err != nil {
		return vol, err
	}
	defer done(&err)

	rc   io.ReadCloser
	done chan struct{}
}

func (c *closeNotifier) Read(p []byte) (n int, err error) {
	n, err = c.rc.Read(p)
	if err != nil {
		if c.done != nil {
			xioutil.SafeClose(c.done)
			c.done = nil
		}
	}
	return n, err
}

func (c *closeNotifier) Close() error {
	if c.done != nil {
		xioutil.SafeClose(c.done)
		c.done = nil
	}
	return c.rc.Close()
}

				xfer := make([]expireObjInfo, len(toDel))
				copy(xfer, toDel)

				var done bool
				select {
				case <-ctx.Done():
					done = true
				case expireCh <- xfer:
					toDel = toDel[:0] // resetting toDel
				}
				if done {
					break
				}
			}
			var match BatchJobExpireFilter
			var found bool
			for _, rule := range r.Rules {

                buf.append('\t').append("action:done");
                buf.append('\t').append("error:").append(escapeValue(e.getLocalizedMessage()).replaceAll("\\s", " "));
                log(buf);
            }
        });
    }

    protected void printStats(final Object keyObj, final StatsObject data, final long begin, final boolean done) {
        final StringBuilder buf = createStringBuffer(keyObj, begin);

			}()
			return
		}
		select {
		case <-m.ctx.Done():
			if debugPrint {
				fmt.Println("Client sending disconnect to mux", m.MuxID)
			}
			m.addErrorNonBlockingClose(internalResp, context.Cause(m.ctx))
			errState = true
			continue
		case req, ok := <-requests:
			if !ok {
				// Done send EOF
				if debugPrint {
					fmt.Println("Client done, sending EOF to mux", m.MuxID)
				}
				msg := message{

		"",
	).Return(nil)

	server := &meshDataplane{
		kubeClient: client.Kube(),
		netServer:  fs,
	}

	handlers := setupHandlers(ctx, client, server, "istio-system")
	client.RunAndWait(ctx.Done())
	go handlers.Start()

	// label the namespace
	labelsPatch := []byte(fmt.Sprintf(`{"metadata":{"labels":{"%s":"%s"}}}`,
		constants.DataplaneModeLabel, constants.DataplaneModeAmbient))

  while read -r value; do
    CLUSTER_NAMES+=("$value")
  done < <(echo "${KUBE_CLUSTERS}" | jq -r '.cluster_name // .clusterName')

  while read -r value; do
    CLUSTER_POD_SUBNETS+=("$value")
  done < <(echo "${KUBE_CLUSTERS}" | jq -r '.pod_subnet // .podSubnet')

  while read -r value; do
    CLUSTER_SVC_SUBNETS+=("$value")
  done < <(echo "${KUBE_CLUSTERS}" | jq -r '.svc_subnet // .svcSubnet')

					done = true
				}
			}
		case <-ctx.Done():
			// Capture timedout locks as failed or took too long
			locksFailed++
			if locksFailed > tolerance {
				// We know that we are not going to get the lock anymore,
				// so exit out and release any locks that did get acquired
				done = true
			}
		}

		if done {
			break
		}
	}

		wg.Add(1)
		go func(ctx context.Context, peerChannel <-chan MetricV2, wg *sync.WaitGroup) {
			defer wg.Done()
			for {
				select {
				case m, ok := <-peerChannel:
					if !ok {
						return
					}
					select {
					case ch <- m:
					case <-ctx.Done():
						return
					}
				case <-ctx.Done():
					return
				}
			}
		}(ctx, peerChannels[index], &wg)
	}

		return err
	}
	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricWalkDir, opts.Bucket, opts.BaseDir)
	if err != nil {
		return err
	}
	defer done(&err)

	return p.storage.WalkDir(ctx, opts, wr)
}

// WalkDir will traverse a directory and return all entries found.
// On success a meta cache stream will be returned, that should be closed when done.