<commit_before>/*
Copyright 2017 Mirantis

Licensed 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.

Based on pkg/kubelet/remote/remote_runtime.go and pkg/kubelet/remote/remote_image.go
from Kubernetes project.
Original copyright notice follows:

Copyright 2016 The Kubernetes Authors.

Licensed 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 criproxy

import (
	"errors"
	"fmt"
	"net"
	"os"
	"os/exec"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/davecgh/go-spew/spew"
	"github.com/golang/glog"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
)

type clientState int

const (
	targetRuntimeAnnotationKey = "kubernetes.io/target-runtime"
	// FIXME: make the following configurable
	// connect timeout when waiting for the socket to become available
	connectWaitTimeout = 500 * time.Millisecond
	clientStateOffline = clientState(iota)
	clientStateConnecting
	clientStateConnected
)

type apiClient struct {
	sync.Mutex
	runtimeapi.RuntimeServiceClient
	runtimeapi.ImageServiceClient
	conn              *grpc.ClientConn
	addr              string
	id                string
	connectionTimeout time.Duration
	state             clientState
	connectErrChs     []chan error
}

func newApiClient(addr string, connectionTimeout time.Duration) *apiClient {
	id := ""
	parts := strings.SplitN(addr, ":", 2)
	if len(parts) == 2 {
		id, addr = parts[0], parts[1]
	}
	return &apiClient{addr: addr, id: id, connectionTimeout: connectionTimeout}
}

func (c *apiClient) isPrimary() bool {
	return c.id == ""
}

func (c *apiClient) currentState() clientState {
	c.Lock()
	defer c.Unlock()
	return c.state
}

func (c *apiClient) connectNonLocked() chan error {
	if c.state == clientStateConnected {
		errCh := make(chan error, 1)
		errCh <- nil
		return errCh
	}

	errCh := make(chan error, 1)
	c.connectErrChs = append(c.connectErrChs, errCh)
	if c.state == clientStateConnecting {
		return errCh
	}

	c.state = clientStateConnecting
	go func() {
		glog.V(1).Infof("Connecting to runtime service %s", c.addr)
		if err := waitForSocket(c.addr, -1, func() error {
			var err error
			c.conn, err = grpc.Dial(c.addr, grpc.WithInsecure(), grpc.WithTimeout(c.connectionTimeout), grpc.WithDialer(dial))
			if err == nil {
				ctx, _ := context.WithTimeout(context.Background(), c.connectionTimeout)
				_, err := runtimeapi.NewRuntimeServiceClient(c.conn).Version(ctx, &runtimeapi.VersionRequest{})
				if err != nil {
					c.conn.Close()
					c.conn = nil
				}
			}
			return err
		}); err != nil {
			glog.Errorf("Failed to find the socket: %v", err)
			errCh <- fmt.Errorf("failed to find the socket: %v", err)
			return
		}

		c.Lock()
		defer c.Unlock()
		glog.V(1).Infof("Connected to runtime service %s", c.addr)
		c.RuntimeServiceClient = runtimeapi.NewRuntimeServiceClient(c.conn)
		c.ImageServiceClient = runtimeapi.NewImageServiceClient(c.conn)
		c.state = clientStateConnected

		errCh <- nil
	}()
	return errCh
}

func (c *apiClient) connect() chan error {
	c.Lock()
	defer c.Unlock()
	return c.connectNonLocked()
}

func (c *apiClient) stopNonLocked() {
	if c.conn == nil {
		return
	}
	if err := c.conn.Close(); err != nil {
		glog.Errorf("Failed to close gRPC connection: %v", err)<commit_msg>Fix incorrect CRI connection error handling<commit_after>/*
Copyright 2017 Mirantis

Licensed 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.

Based on pkg/kubelet/remote/remote_runtime.go and pkg/kubelet/remote/remote_image.go
from Kubernetes project.
Original copyright notice follows:

Copyright 2016 The Kubernetes Authors.

Licensed 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 criproxy

import (
	"errors"
	"fmt"
	"net"
	"os"
	"os/exec"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/davecgh/go-spew/spew"
	"github.com/golang/glog"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
)

type clientState int

const (
	targetRuntimeAnnotationKey = "kubernetes.io/target-runtime"
	// FIXME: make the following configurable
	// connect timeout when waiting for the socket to become available
	connectWaitTimeout = 500 * time.Millisecond
	clientStateOffline = clientState(iota)
	clientStateConnecting
	clientStateConnected
)

type apiClient struct {
	sync.Mutex
	runtimeapi.RuntimeServiceClient
	runtimeapi.ImageServiceClient
	conn              *grpc.ClientConn
	addr              string
	id                string
	connectionTimeout time.Duration
	state             clientState
	connectErrChs     []chan error
}

func newApiClient(addr string, connectionTimeout time.Duration) *apiClient {
	id := ""
	parts := strings.SplitN(addr, ":", 2)
	if len(parts) == 2 {
		id, addr = parts[0], parts[1]
	}
	return &apiClient{addr: addr, id: id, connectionTimeout: connectionTimeout}
}

func (c *apiClient) isPrimary() bool {
	return c.id == ""
}

func (c *apiClient) currentState() clientState {
	c.Lock()
	defer c.Unlock()
	return c.state
}

func (c *apiClient) connectNonLocked() chan error {
	if c.state == clientStateConnected {
		errCh := make(chan error, 1)
		errCh <- nil
		return errCh
	}

	errCh := make(chan error, 1)
	c.connectErrChs = append(c.connectErrChs, errCh)
	if c.state == clientStateConnecting {
		return errCh
	}

	c.state = clientStateConnecting
	go func() {
		glog.V(1).Infof("Connecting to runtime service %s", c.addr)
		if err := waitForSocket(c.addr, -1, func() error {
			var err error
			c.conn, err = grpc.Dial(c.addr, grpc.WithInsecure(), grpc.WithTimeout(c.connectionTimeout), grpc.WithDialer(dial))
			if err == nil {
				ctx, _ := context.WithTimeout(context.Background(), c.connectionTimeout)
				_, err := runtimeapi.NewRuntimeServiceClient(c.conn).Version(ctx, &runtimeapi.VersionRequest{})
				if err != nil {
					c.conn.Close()
					c.conn = nil
				}
			}
			return err
		}); err != nil {
			glog.Errorf("Failed to find the socket: %v", err)
			err = fmt.Errorf("failed to find the socket: %v", err)
			for _, ch := range c.connectErrChs {
				ch <- err
			}
			return
		}

		c.Lock()
		defer c.Unlock()
		glog.V(1).Infof("Connected to runtime service %s", c.addr)
		c.RuntimeServiceClient = runtimeapi.NewRuntimeServiceClient(c.conn)
		c.ImageServiceClient = runtimeapi.NewImageServiceClient(c.conn)
		c.state = clientStateConnected

		for _, ch := range c.connectErrChs {
			ch <- nil
		}
	}()
	return errCh
}

func (c *apiClient) connect() chan error {
	c.Lock()
	defer c.Unlock()
	return c.connectNonLocked()
}

func (c *apiClient) stopNonLocked() {
	if c.conn == nil {
		return
	}
	if err := c.conn.Close(); err != nil {
		glog.Errorf("Failed to close gRPC connection: %v", err)
{'fraction_non_alphanumeric': 0.10590418744925183, 'fraction_numerical': 0.007191741097320496, 'mean_word_length': 3.8766968325791855, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5769200', 'n_tokens_mistral': 2946, 'n_tokens_neox': 2730, 'n_words': 860}