run_test.go

// Copyright 2021 by Leipzig University Library, http://ub.uni-leipzig.de
//                   The Finc Authors, http://finc.info
//                   Martin Czygan, <martin.czygan@uni-leipzig.de>
//
// This file is part of some open source application.
//
// Some open source application is free software: you can redistribute
// it and/or modify it under the terms of the GNU General Public
// License as published by the Free Software Foundation, either
// version 3 of the License, or (at your option) any later version.
//
// Some open source application is distributed in the hope that it will
// be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with Foobar.  If not, see <http://www.gnu.org/licenses/>.
//
// @license GPL-3.0+ <http://spdx.org/licenses/GPL-3.0+>

package esbulk

import (
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/url"
	"os"
	"os/exec"
	"os/user"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/docker/docker/api/types/container"
	"github.com/segmentio/encoding/json"
	"github.com/sethgrid/pester"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/wait"
)

func TestIncompleteConfig(t *testing.T) {
	skipNoDocker(t)
	var cases = []struct {
		help string
		r    Runner
		err  error
	}{
		{help: "no default index name", r: Runner{}, err: ErrNoWorkers},
		{help: "no such host", r: Runner{
			IndexName:  "abc",
			BatchSize:  10,
			NumWorkers: 1,
			Servers:    []string{"http://broken.server:9200"},
		}, err: &url.Error{Op: "Get", URL: "http://broken.server:9200/abc"}},
	}
	for _, c := range cases {
		err := c.r.Run()
		switch err.(type) {
		case nil:
			if c.err != nil {
				t.Fatalf("got: %#v, %T, want: %v [%s]", err, err, c.err, c.help)
			}
		case *url.Error:
			// For now, only check whether we expect an error.
			if c.err == nil {
				t.Fatalf("got: %#v, %T, want: %v [%s]", err, err, c.err, c.help)
			}
		default:
			if err != c.err {
				t.Fatalf("got: %#v, %T, want: %v [%s]", err, err, c.err, c.help)
			}
		}
	}
}

// startServer starts an elasticsearch server from image, exposing the http
// port. Note that the Java heap required may be 2GB or more.
func startServer(ctx context.Context, image string, httpPort int) (testcontainers.Container, error) {
	var (
		hp    = fmt.Sprintf("%d:9200/tcp", httpPort)
		parts = strings.Split(image, ":")
		tag   string
	)
	if len(parts) == 2 {
		tag = parts[1]
	} else {
		tag = "latest"
	}
	var (
		name = fmt.Sprintf("esbulk-test-es-%s-%d", tag, time.Now().UnixNano())
		req  = testcontainers.ContainerRequest{
			Image: image,
			Name:  name,
			Env: map[string]string{
				"discovery.type": "single-node",
				// If you’re starting a single-node Elasticsearch cluster in a
				// Docker container, security will be automatically enabled and
				// configured for you. -- https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html#docker-cli-run-dev-mode
				"xpack.security.enabled": "false",
				"ES_JAVA_OPTS":           "-Xms4g -Xmx4g",
			},
			ExposedPorts: []string{hp},
			WaitingFor:   wait.ForLog("started"),
			// $ docker run --platform linux/amd64 m 8g -e ES_JAVA_OPTS="-Xms512m -Xmx512m" elasticsearch:2.3.4
			// library initialization failed - unable to allocate file descriptor table - out of memory#
			HostConfigModifier: func(hc *container.HostConfig) {
				hc.CgroupnsMode = container.CgroupnsModeHost
				hc.Resources = container.Resources{
					Ulimits: []*container.Ulimit{{
						Name: "nofile",
						Hard: 65536,
						Soft: 65536,
					}},
				}
			},
		}
	)
	return testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
		ProviderType:     testcontainers.ProviderDefault,
		ContainerRequest: req,
		Started:          true,
	})
}

// logReader reads data from reader and bot logs it and returns it. Fails, if
// reading fails.
func logReader(t *testing.T, r io.Reader) []byte {
	b, err := ioutil.ReadAll(r)
	if err != nil {
		t.Fatalf("read failed: %s", err)
		return nil
	}
	t.Logf("%s", string(b))
	return b
}

// skipNoDocker skips a test, if docker is not running. Also support podman.
func skipNoDocker(t *testing.T) {
	noDocker := false
	cmd := exec.Command("systemctl", "is-active", "docker")
	b, err := cmd.CombinedOutput()
	if err != nil {
		noDocker = true
	}
	if strings.TrimSpace(string(b)) != "active" {
		noDocker = true
	}
	if !noDocker {
		// We found some docker.
		return
	}
	// Otherwise, try podman.
	_, err = exec.LookPath("podman")
	if err == nil {
		t.Logf("podman detected")
		// DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock
		usr, err := user.Current()
		if err != nil {
			t.Logf("cannot get UID, set DOCKER_HOST manually")
		} else {
			sckt := fmt.Sprintf("unix:///run/user/%v/podman/podman.sock", usr.Uid)
			os.Setenv("DOCKER_HOST", sckt)
			t.Logf("set DOCKER_HOST to %v", sckt)
		}
		noDocker = false
	}
	if noDocker {
		t.Skipf("docker not installed or not running")
	}
}

func TestMinimalConfig(t *testing.T) {
	skipNoDocker(t)
	ctx := context.Background()
	var imageConf = []struct {
		ElasticsearchMajorVersion int
		Image                     string
		HttpPort                  int
	}{
		// $ docker pull elasticsearch:2.3.4
		// 2.3.4: Pulling from library/elasticsearch
		// ca69fe441e9d: Already exists
		// failed to unpack image on snapshotter overlayfs: apply layer error
		// for "docker.io/library/elasticsearch:2.3.4": failed to extract layer
		// sha256:2f71b45e4e254ddceb187b1467f5471f0e14d7124ac2dd7fdd7ddbc76e13f0e5:
		// NotFound: failed to get reader from content store: content digest
		// sha256:357ea8c3d80bc25792e010facfc98aee5972ebc47e290eb0d5aea3671a901cab:
		// not found
		// {2, "elasticsearch:2.3.4", 39200},
		{5, "elasticsearch:5.6.16", 39200},
		{6, "elasticsearch:6.8.14", 39200},
		{7, "elasticsearch:7.17.7", 39200}, // https://is.gd/MPwhaM, https://is.gd/RJ4LOZ, ...
		{8, "elasticsearch:8.6.0", 39200},
	}
	log.Printf("testing %d versions: %v", len(imageConf), imageConf)
	for _, conf := range imageConf {
		wrapper := func() error {
			c, err := startServer(ctx, conf.Image, conf.HttpPort)
			if err != nil {
				t.Fatalf("could not start test container for %v: %v", conf.Image, err)
			}
			defer func() {
				if err := c.Terminate(ctx); err != nil {
					t.Errorf("could not kill container: %v", err)
				}
			}()
			base := fmt.Sprintf("http://localhost:%d", conf.HttpPort)
			resp, err := pester.Get(base)
			if err != nil {
				t.Fatalf("request failed: %v", err)
			}
			defer resp.Body.Close()
			logReader(t, resp.Body)
			t.Logf("server should be up at %s", base)

			var cases = []struct {
				filename  string
				indexName string
				numDocs   int64
				err       error
			}{
				{"fixtures/v10k.jsonl", "abc", 10000, nil},
			}
			for _, c := range cases {
				f, err := os.Open(c.filename)
				if err != nil {
					return fmt.Errorf("could not open fixture: %s", c.filename)
				}
				defer f.Close()
				r := Runner{
					Servers:         []string{"http://localhost:39200"},
					BatchSize:       5000,
					NumWorkers:      1,
					RefreshInterval: "1s",
					IndexName:       "abc",
					File:            f,
					Verbose:         true,
				}
				if conf.ElasticsearchMajorVersion < 7 {
					r.DocType = "any" // deprecated with ES7, fails with ES8
				}
				err = r.Run()
				if err != c.err {
					return fmt.Errorf("got %v, want %v", err, c.err)
				}
				searchURL := fmt.Sprintf("%s/%s/_search", base, r.IndexName)
				resp, err = pester.Get(searchURL)
				if err != nil {
					return fmt.Errorf("could not query es: %v", err)
				}
				defer resp.Body.Close()
				b := logReader(t, resp.Body)
				var (
					sr7 SearchResponse7
					sr6 SearchResponse6
				)
				if err = json.Unmarshal(b, &sr7); err != nil {
					if err = json.Unmarshal(b, &sr6); err != nil {
						t.Errorf("could not parse json response (6, 7): %v", err)
					} else {
						t.Log("es6 detected")
					}
				}
				if sr7.Hits.Total.Value != c.numDocs && sr6.Hits.Total != c.numDocs {
					t.Errorf("expected %d docs", c.numDocs)
				}
			}
			// Save logs.
			rc, err := c.Logs(ctx)
			if err != nil {
				log.Printf("logs not available: %v", err)
			}
			if err := os.MkdirAll("logs", 0755); err != nil {
				if !os.IsExist(err) {
					log.Printf("create dir failed: %v", err)
				}
			}
			cname, err := c.Name(ctx)
			if err != nil {
				t.Logf("failed to get container name: %v", err)
			}
			fn := fmt.Sprintf("logs/%s-%s.log", time.Now().Format("20060102150405"), strings.TrimLeft(cname, "/"))
			f, err := os.Create(fn)
			if err != nil {
				log.Printf("failed to create log file: %v", err)
			}
			defer f.Close()
			log.Printf("logging to %s", fn)
			if _, err := io.Copy(f, rc); err != nil {
				log.Printf("log failed: %v", err)
			}
			return nil
		}
		if err := wrapper(); err != nil {
			t.Fatal(err)
		}
	}
}

func TestGH32(t *testing.T) {
	skipNoDocker(t)
	ctx := context.Background()
	// 7.17.7 bundles a Java version, that has improved cgroups2 support
	c, err := startServer(ctx, "elasticsearch:7.17.7", 39200)
	if err != nil {
		t.Fatalf("could not start test container: %v", err)
	}
	defer func() {
		if err := c.Terminate(ctx); err != nil {
			t.Errorf("could not kill container: %v", err)
		}
	}()
	base := fmt.Sprintf("http://localhost:%d", 39200)
	resp, err := pester.Get(base)
	if err != nil {
		t.Fatalf("request failed: %v", err)
	}
	defer resp.Body.Close()
	logReader(t, resp.Body)
	t.Logf("server should be up at %s", base)

	f, err := os.Open("fixtures/v10k.jsonl")
	if err != nil {
		t.Errorf("could not open fixture: %v", err)
	}
	defer f.Close()
	r := Runner{
		Servers:         []string{"http://localhost:39200"},
		BatchSize:       5000,
		NumWorkers:      1,
		RefreshInterval: "1s",
		Mapping:         `{}`,
		IndexName:       "abc",
		DocType:         "any", // deprecated with ES7
		File:            f,
		Verbose:         true,
	}
	// this should fail with #32
	err = r.Run()
	if err != nil {
		t.Logf("expected err: %v", err)
	} else {
		t.Fatalf("expected fail, see #32")
	}
	// w/o doctype, we should be good
	r = Runner{
		Servers:         []string{"http://localhost:39200"},
		BatchSize:       5000,
		NumWorkers:      1,
		RefreshInterval: "1s",
		Mapping:         `{}`,
		IndexName:       "abc",
		File:            f,
		Verbose:         true,
	}
	err = r.Run()
	if err != nil {
		t.Fatalf("unexpected failure: %v", err)
	}
}

type SearchResponse6 struct {
	Hits struct {
		Hits []struct {
			Id     string  `json:"_id"`
			Index  string  `json:"_index"`
			Score  float64 `json:"_score"`
			Source struct {
				V string `json:"v"`
			} `json:"_source"`
			Type string `json:"_type"`
		} `json:"hits"`
		MaxScore float64 `json:"max_score"`
		Total    int64   `json:"total"`
	} `json:"hits"`
	Shards struct {
		Failed     int64 `json:"failed"`
		Skipped    int64 `json:"skipped"`
		Successful int64 `json:"successful"`
		Total      int64 `json:"total"`
	} `json:"_shards"`
	TimedOut bool  `json:"timed_out"`
	Took     int64 `json:"took"`
}

type SearchResponse7 struct {
	Hits struct {
		Hits []struct {
			Id     string  `json:"_id"`
			Index  string  `json:"_index"`
			Score  float64 `json:"_score"`
			Source struct {
				V string `json:"v"`
			} `json:"_source"`
			Type string `json:"_type"`
		} `json:"hits"`
		MaxScore float64 `json:"max_score"`
		Total    struct {
			Relation string `json:"relation"`
			Value    int64  `json:"value"`
		} `json:"total"`
	} `json:"hits"`
	Shards struct {
		Failed     int64 `json:"failed"`
		Skipped    int64 `json:"skipped"`
		Successful int64 `json:"successful"`
		Total      int64 `json:"total"`
	} `json:"_shards"`
	TimedOut bool  `json:"timed_out"`
	Took     int64 `json:"took"`
}

func TestContextCancellation(t *testing.T) {
	tests := []struct {
		name        string
		cancelDelay time.Duration
		expectError bool
	}{
		{"early cancellation", 10 * time.Millisecond, true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
			defer cancel()

			if tt.cancelDelay > 0 {
				time.AfterFunc(tt.cancelDelay, cancel)
			}
			options := Options{
				BatchSize:      10,
				Verbose:        false, // Reduce noise in test output
				Servers:        []string{"http://localhost:9200"},
				Index:          "test-index",
				RequestTimeout: 5 * time.Second,
			}
			var (
				lines   = make(chan string, 100)
				errChan = make(chan error, 1)
				wg      sync.WaitGroup
			)
			wg.Add(1)
			for i := 0; i < 5; i++ {
				lines <- `{"test": "data"}`
			}
			close(lines)
			err := Worker(ctx, "test-worker", options, lines, &wg, errChan)
			if tt.expectError {
				select {
				case <-ctx.Done():
					t.Logf("Context cancelled as expected: %v", ctx.Err())
				case err := <-errChan:
					// May get connection errors, which is ok when context is cancelled
					t.Logf("worker error (expected): %v", err)
				case <-time.After(200 * time.Millisecond):
					t.Error("expected context cancellation but worker didn't finish")
				}
			} else {
				if err != nil {
					t.Errorf("worker failed unexpectedly: %v", err)
				}
			}
		})
	}
}

func TestWorkerRespectsImmediateCancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	options := Options{
		BatchSize:      10,
		Verbose:        false,
		Servers:        []string{"http://localhost:9200"},
		Index:          "test-index",
		RequestTimeout: 30 * time.Second,
	}
	var (
		lines   = make(chan string, 100)
		errChan = make(chan error, 1)
	)
	var wg sync.WaitGroup
	wg.Add(1)
	var (
		start = time.Now()
		err   = Worker(ctx, "test-worker", options, lines, &wg, errChan)
	)
	elapsed := time.Since(start)
	if elapsed > 100*time.Millisecond {
		t.Errorf("worker should return immediately when context is cancelled, but took %v", elapsed)
	}
	if err != nil {
		t.Errorf("unexpected error: %v", err)
	}
}

func TestCreateHTTPRequestWithContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	options := Options{
		Username: "testuser",
		Password: "testpass",
	}
	req, err := CreateHTTPRequestWithContext(ctx, "GET", "http://example.com", nil, options)
	if err != nil {
		t.Fatalf("failed to create request: %v", err)
	}
	if req.Method != "GET" {
		t.Errorf("expected method GET, got %s", req.Method)
	}
	if req.URL.String() != "http://example.com" {
		t.Errorf("expected URL http://example.com, got %s", req.URL.String())
	}
	username, password, ok := req.BasicAuth()
	if !ok {
		t.Error("expected basic auth to be set")
	}

	if username != "testuser" || password != "testpass" {
		t.Errorf("expected basic auth user:testuser pass:testpass, got %s:%s", username, password)
	}
}

func TestCreateHTTPClientWithTimeout(t *testing.T) {
	tests := []struct {
		name            string
		insecureSkip    bool
		timeout         time.Duration
		expectedTimeout time.Duration
	}{
		{"default timeout", false, 0, 30 * time.Second},
		{"custom timeout", false, 45 * time.Second, 45 * time.Second},
		{"insecure with timeout", true, 60 * time.Second, 60 * time.Second},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			client := CreateHTTPClient(tt.insecureSkip, tt.timeout)
			if client.Timeout != tt.expectedTimeout {
				t.Errorf("expected timeout %v, got %v", tt.expectedTimeout, client.Timeout)
			}
		})
	}
}