You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: datacatalog.go
    path of file: ./repos/flyte/flytepropeller/pkg/controller/nodes/catalog/datacatalog
    
    the code of the file until where you have to start completion: package datacatalog

import (
	"context"
	"crypto/x509"
	"fmt"
	"time"

	"github.com/golang/protobuf/ptypes"
	grpcRetry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pkg/errors"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/otel/propagation"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/status"
	"k8s.io/apimachinery/pkg/util/uuid"

	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/datacatalog"
	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/catalog"
	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/io"
	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/ioutils"
	"github.com/flyteorg/flyte/flytestdlib/logger"
	"github.com/flyteorg/flyte/flytestdlib/otelutils"
)

var (
	_ catalog.Client = &CatalogClient{}
)

// CatalogClient is the client that caches task executions to DataCatalog service.
type CatalogClient struct {
	client      datacatalog.DataCatalo
    
func (m *CatalogClient) Get(ctx context.Context, key catalog.Key) (catalog.Entry, error) {
	dataset, err := m.GetDataset(ctx, key)
	if err != nil {
		logger.Debugf(ctx, "DataCatalog failed to get dataset for ID %s, err: %+v", key.Identifier.String(), err)
		return catalog.Entry{}, errors.Wrapf(err, "DataCatalog failed to get dataset for ID %s", key.Identifier.String())
	}

	inputs := &core.LiteralMap{}
	if key.TypedInterface.Inputs != nil {
		retInputs, err := key.InputReader.Get(ctx)
		if err != nil {
			return catalog.Entry{}, errors.Wrap(err, "failed to read inputs when trying to query catalog")
		}
		inputs = retInputs
	}

	tag, err := GenerateArtifactTagName(ctx, inputs, key.CacheIgnoreInputVars)
	if err != nil {
		logger.Errorf(ctx, "DataCatalog failed to generate tag for inputs %+v, err: %+v", inputs, err)
		return catalog.Entry{}, err
	}

	artifact, err := m.GetArtifactByTag(ctx, tag, dataset)
	if err != nil {
		logger.Debugf(ctx, "DataCatalog failed to get artifact by tag %+v, err: %+v", tag, err)
		return catalog.Entry{}, err
	}
	logger.Debugf(ctx, "Artifact found %v from tag %v", artifact, tag)

	var relevantTag *datacatalog.Tag
	if len(artifact.GetTags()) > 0 {
		// TODO should we look through all the tags to find the relevant one?
		relevantTag = artifact.GetTags()[0]
	}

	source, err := GetSourceFromMetadata(dataset.GetMetadata(), artifact.GetMetadata(), key.Identifier)
	if err != nil {
		return catalog.Entry{}, fmt.Errorf("failed to get source from metadata. Error: %w", err)
	}

	md := EventCatalogMetadata(dataset.GetId(), relevantTag, source)

	outputs, err := GenerateTaskOutputsFromArtifact(key.Identifier, key.TypedInterface, artifact)
	if err != nil {
		logger.Errorf(ctx, "DataCatalog failed to get outputs from artifact %+v, err: %+v", artifact.Id, err)
		return catalog.NewCatalogEntry(ioutils.NewInMemoryOutputReader(outputs, nil, nil), catalog.NewStatus(core.CatalogCacheStatus_CACHE_MISS, md)), err
	}

	logger.Infof(ctx, "Retrieved %v outputs from artifact %v, tag: %v", len(outputs.Literals), artifact.Id, tag)
	return catalog.NewCatalogEntry(ioutils.NewInMemoryOutputReader(outputs, nil, nil), catalog.NewStatus(core.CatalogCacheStatus_CACHE_HIT, md)), nil
}