В этой статье я хочу поделиться примером, как можно развернуть простое приложение на Golang в Kubernetes, с помощью helm чартов и skaffold скриптов. Думаю, данная статья может быть полезной тем разработчикам, которые только знакомятся с Kubernetes, а возможно и более опытным разработчикам, которые смогут почерпнуть что то интересное для себя.
Хочу заранее отметить, что мое приложение носит исключительно демонстрационный характер, поэтому весь исходный код в одном main файле:
cmd/app/main.go
package main
import (
"context"
"database/sql"
"errors"
"flag"
"fmt"
"log"
"net"
"net/http"
"os"
"os/signal"
"syscall"
"time"
apiv1 "k8-golang-demo/gen/pb-go/com.example/usersvcapi/v1"
"github.com/google/uuid"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
run "github.com/oklog/oklog/pkg/group"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
// justifying it
_ "github.com/go-sql-driver/mysql"
)
func main() {
// Flags.
//
fs := flag.NewFlagSet("", flag.ExitOnError)
grpcAddr := fs.String("grpc-addr", ":6565", "grpc address")
httpAddr := fs.String("http-addr", ":8080", "http address")
if err := fs.Parse(os.Args[1:]); err != nil {
log.Fatal(err)
}
// Setup database.
//
db, err := NewDatabase(
os.Getenv("DATABASE_DRIVER"),
os.Getenv("DATABASE_NAME"),
os.Getenv("DATABASE_USERNAME"),
os.Getenv("DATABASE_PASSWORD"),
os.Getenv("DATABASE_HOST"),
os.Getenv("DATABASE_PORT"),
)
if err != nil {
log.Fatal(err)
}
conn := db.GetConnection()
defer func() {
_ = db.CloseConnection()
}()
// Setup gRPC servers.
//
baseGrpcServer := grpc.NewServer()
userGrpcServer := NewUserGRPCServer(conn, "users")
apiv1.RegisterUserServiceServer(baseGrpcServer, userGrpcServer)
// Setup gRPC gateway.
//
ctx := context.Background()
rmux := runtime.NewServeMux()
mux := http.NewServeMux()
mux.Handle("/", rmux)
{
err := apiv1.RegisterUserServiceHandlerServer(ctx, rmux, userGrpcServer)
if err != nil {
log.Fatal(err)
}
}
// Serve.
//
var g run.Group
{
grpcListener, err := net.Listen("tcp", *grpcAddr)
if err != nil {
log.Fatal(err)
}
g.Add(func() error {
log.Printf("Serving grpc address %s", *grpcAddr)
return baseGrpcServer.Serve(grpcListener)
}, func(error) {
grpcListener.Close()
})
}
{
httpListener, err := net.Listen("tcp", *httpAddr)
if err != nil {
log.Fatal(err)
}
g.Add(func() error {
log.Printf("Serving http address %s", *httpAddr)
return http.Serve(httpListener, mux)
}, func(err error) {
httpListener.Close()
})
}
{
cancelInterrupt := make(chan struct{})
g.Add(func() error {
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
select {
case sig := <-c:
return fmt.Errorf("received signal %s", sig)
case <-cancelInterrupt:
return nil
}
}, func(error) {
close(cancelInterrupt)
})
}
if err := g.Run(); err != nil {
log.Fatal(err)
}
}
type userServer struct {
conn *sql.DB
tableName string
}
func NewUserGRPCServer(conn *sql.DB, tableName string) apiv1.UserServiceServer {
return &userServer{
conn: conn,
tableName: tableName,
}
}
func (s *userServer) CreateUser(ctx context.Context, req *apiv1.CreateUserRequest) (*apiv1.CreateUserResponse, error) {
if req.User == nil {
return nil,
status.Error(codes.InvalidArgument, "User required")
}
id, err := uuid.NewRandom()
if err != nil {
return nil,
status.Error(codes.Internal, err.Error())
}
query := `INSERT INTO ` + s.tableName + `(id, name, type) VALUES(?,?,?)`
_, err = runWriteTransaction(s.conn, query, id, req.User.Name, req.User.Type)
if err != nil {
return nil, err
}
return &apiv1.CreateUserResponse{
Id: id.String(),
}, nil
}
func (s *userServer) GetUsers(ctx context.Context, req *apiv1.GetUsersRequest) (*apiv1.GetUsersResponse, error) {
query := `SELECT id, name, type FROM ` + s.tableName + ``
users := []*apiv1.UserRead{}
err := runQuery(s.conn, query, []interface{}{}, func(rows *sql.Rows) error {
found := apiv1.UserRead{}
err := rows.Scan(
&found.Id,
&found.Name,
&found.Type,
)
if err != nil {
return err
}
users = append(users, &found)
return nil
})
if err != nil {
return nil,
status.Error(codes.NotFound, fmt.Errorf("User not found, err %v", err).Error())
}
return &apiv1.GetUsersResponse{
Users: users,
}, nil
}
// SQLDatabase is the interface that provides sql methods.
type SQLDatabase interface {
GetConnection() *sql.DB
CloseConnection() error
}
type db struct {
conn *sql.DB
}
// NewDatabase creates a new sql database connection with the base migration setup.
func NewDatabase(driver, database, username, password, host string, port string) (SQLDatabase, error) {
source := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", username, password, host, port, database)
sqlconn, err := sql.Open(driver, source)
if err != nil {
return nil, fmt.Errorf("Failed to open database connection, err:%v", err)
}
return &db{
conn: sqlconn,
}, nil
}
func (h *db) GetConnection() *sql.DB {
return h.conn
}
func (h *db) CloseConnection() error {
if h.conn == nil {
return errors.New("Cannot close the connection because the connection is nil")
}
return h.conn.Close()
}
func runWriteTransaction(db *sql.DB, query string, args ...interface{}) (sql.Result, error) {
ctx, stop := context.WithCancel(context.Background())
defer stop()
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelDefault})
if err != nil {
return nil, err
}
defer func() {
_ = tx.Rollback() // The rollback will be ignored if the tx has been committed later in the function.
}()
stmt, err := tx.Prepare(query)
if err != nil {
return nil, err
}
defer stmt.Close() // Prepared statements take up server resources and should be closed after use.
queryResult, err := stmt.Exec(args...)
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return queryResult, err
}
func runQuery(db *sql.DB, query string, args []interface{}, f func(*sql.Rows) error) error {
ctx, stop := context.WithCancel(context.Background())
defer stop()
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
queryResult, err := db.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer func() {
_ = queryResult.Close()
}()
for queryResult.Next() {
err = f(queryResult)
if err != nil {
return err
}
}
return nil
}
Чтобы приложение было готово к запуску в Kubernetes, нам необходимо контейнеризировать его. Опишем докер файл для нашего сервиса:
Dockerfile
FROM golang:1.16-alpine as builder
WORKDIR /build
COPY go.mod .
COPY go.sum .
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o /main cmd/app/main.go
FROM alpine:3
COPY --from=builder main /bin/main
ENTRYPOINT ["/bin/main"]
Опишем код для миграции данных, которые на входе будут принимать SQL файлы и выполнять их в соответствующем порядке. В качестве фреймворка мы будем использовать достаточно популярный инструмент golang-migrate:
cmd/migrations/main.go
package main
import (
"fmt"
"os"
"database/sql"
"github.com/cenk/backoff"
_ "github.com/go-sql-driver/mysql"
"github.com/golang-migrate/migrate/v4"
"github.com/golang-migrate/migrate/v4/database/mysql"
_ "github.com/golang-migrate/migrate/v4/source/file"
)
const migrationsPath = "file://migrations"
func main() {
// Open connection.
conn, driver := getConnection()
// Close connection.
defer func() {
if conn != nil {
_ = conn.Close()
}
}()
// Establish connection to a database.
establishConnectionWithRetry(conn)
// Starting migration job.
err := migrateSQL(conn, driver)
if err != nil {
panic(err)
} else {
fmt.Println("Migration successfully finished.")
}
}
func getConnection() (*sql.DB, string) {
driver := os.Getenv("MYSQL_DRIVER")
address := os.Getenv("MYSQL_HOST") + ":" + os.Getenv("MYSQL_PORT_NUMBER")
username := os.Getenv("MYSQL_DATABASE_USER")
password := os.Getenv("MYSQL_DATABASE_PASSWORD")
database := os.Getenv("MYSQL_DATABASE_NAME")
// Open may just validate its arguments without creating a connection to the database.
sqlconn, err := sql.Open(driver, username+":"+password+"@tcp("+address+")/"+database)
if err != nil {
panic("Cannot establish connection to a database")
}
return sqlconn, driver
}
// This function executes the migration scripts.
func migrateSQL(conn *sql.DB, driverName string) error {
driver, _ := mysql.WithInstance(conn, &mysql.Config{})
m, err := migrate.NewWithDatabaseInstance(
migrationsPath,
driverName,
driver,
)
if err != nil {
return err
}
if err := m.Up(); err != nil && err != migrate.ErrNoChange {
return err
}
return nil
}
func establishConnectionWithRetry(conn *sql.DB) {
b := backoff.NewExponentialBackOff()
// We wait forever until the connection will be established.
// In practice k8s will kill the pod if it takes too long.
b.MaxElapsedTime = 0
_ = backoff.Retry(func() error {
fmt.Println("Connecting to a database ...")
// Ping verifies a connection to the database is still alive,
// establishing a connection if necessary.
if errPing := conn.Ping(); errPing != nil {
return fmt.Errorf("ping failed %v", errPing)
}
return nil
}, b)
}
Теперь когда утилита для миграции данных готова, нам нужно описать сами миграции, для этого в директории migrations/sql создадим:
1_create_table_foobar.up.sql
CREATE TABLE IF NOT EXISTS users (
id VARCHAR(40),
name VARCHAR(255),
type SMALLINT UNSIGNED,
PRIMARY KEY(id)
);
В кластере Kubernetes мы будем запускать миграции данных в качестве джобов, и соответственно нам их также нужно контейнеризировать:
migrations/Dockerfile
FROM golang:1.16-alpine as builder
WORKDIR /build
COPY go.mod .
COPY go.sum .
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o /main cmd/migrations/main.go
FROM alpine:3
COPY --from=builder main /bin/main
COPY --from=builder build/migrations/sql /migrations
ENTRYPOINT ["/bin/main"]
Мы только что создали контейнер для миграции данных и скопировали туда SQL файлы. Наше приложение почти готово к запуску в Kubernetes, самое время описать манифесты.
helm create k8-golang-demo
Отлично, только что мы создали базовый шаблон с определениями ресурсов, необходимые для запуска приложения внутри кластера Kubernetes. Приступим к редактированию ресурсов шаблона под наши нужды.
Chart.yaml - содержит метаданные, общую информацию о чарте, а так же зависимости на другие helm чарты. Поскольку наше приложение использует базу данных MySQL в качестве зависимости, нам необходимо описать эту зависимость в блоке dependencies:
Chart.yaml
Из важного:
values.yaml
# Default values for k8-golang-demo.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.
replicaCount: 1
image:
hostname: docker.io
repository: /golang-enthusiast/k8-golang-demo
tag: 0.1.0
pullPolicy: IfNotPresent
migration:
image:
hostname: docker.io
repository: /golang-enthusiast/k8-golang-demo-data-migration
tag: 0.3.0
pullPolicy: IfNotPresent
service:
type: ClusterIP
protocol: TCP
port: 6565
httpPort: 8080
name: grpc
ingress:
name: http
protocol: HTTP
port: 80
extension: svc.cluster.local
serviceAccount:
create: true
mysql:
enabled: false
host: mysql
mysqlDriver: mysql
mysqlRootPassword: test
mysqlDatabase: test
mysqlUser: admin
mysqlPassword: test
service:
port: 3306
initdbScripts:
initdb.sql: |-
CREATE DATABASE IF NOT EXISTS test DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;
CREATE USER 'admin'@'%' IDENTIFIED BY 'test';
GRANT ALL PRIVILEGES ON *.* TO 'admin'@'%' WITH GRANT OPTION;
FLUSH PRIVILEGES;
primary:
persistence:
enabled: false
storageClass: standard
grpcGateway:
service:
protocol: TCP
port: 8080
name: http
skaffold: false
В значениях мы описали конфигурацию базы данных MySQL. В общем, конфигурация может различаться от поставщика к поставщику helm чартов. Т.к. мы используем стабильные чарты от bitnami, то и конфигурацию мы должны делать соответствующую, которая описана тут.
Так же мы описали образы, которые будут использоваться в нашем приложении. Их всего два, а именно образ для основного приложения и образ для миграции данных. К остальным значениям мы вернемся чуть позже.
deployment.yaml - содержит инструкции для развертывания приложения.
deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ include "k8-golang-demo.fullname" . }}
namespace: {{ .Release.Namespace }}
labels:
{{- include "k8-golang-demo.labels" . | nindent 4 }}
spec:
replicas: {{ .Values.replicaCount }}
selector:
matchLabels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
template:
metadata:
labels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
spec:
containers:
- name: {{ .Chart.Name }}
image: {{ include "k8-golang-demo.image" . }}
imagePullPolicy: {{ .Values.image.pullPolicy }}
env:
- name: DATABASE_HOST
value: "{{ .Release.Name }}-{{ index .Values.mysql.host }}"
- name: DATABASE_PORT
value: {{ .Values.mysql.service.port | quote }}
- name: DATABASE_NAME
value: {{ .Values.mysql.mysqlDatabase | quote }}
- name: DATABASE_USERNAME
value: {{ .Values.mysql.mysqlUser | quote }}
- name: DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
key: mysqlPassword
- name: DATABASE_DRIVER
value: {{ .Values.mysql.mysqlDriver | quote }}
ports:
- name: {{ .Values.service.name }}
containerPort: {{ .Values.service.port }}
protocol: {{ .Values.service.protocol }}
- name: {{ .Values.grpcGateway.service.name }}
containerPort: {{ .Values.grpcGateway.service.port }}
protocol: {{ .Values.grpcGateway.service.protocol }}
В инструкции по развертыванию приложения в секции containers мы описали какой контейнер стоит использовать и с какими переменными окружениями, так же указали какие порты должны быть использованы. Т.к. наше приложение использует два транспортных протокола - gRPC + REST, то следует указать описание обоих в секции ports.
data-migration-job.yaml - здесь мы будем описывать инструкции по развертыванию заданий (джобов) для миграции данных.
data-migration-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
name: {{ include "k8-golang-demo.fullname" . }}-data-migration-job
namespace: {{ .Release.Namespace }}
labels:
{{ include "k8-golang-demo.labels" . | indent 4 }}
annotations:
# This is what defines this resource as a hook. Without this line, the
# job is considered part of the release.
"helm.sh/hook": post-install, post-upgrade
"helm.sh/hook-weight": "-5"
"helm.sh/hook-delete-policy": before-hook-creation
spec:
template:
metadata:
labels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}-migration
app.kubernetes.io/instance: {{ .Release.Name }}
annotations:
readiness.status.sidecar.istio.io/applicationPorts: ""
spec:
restartPolicy: Never
initContainers:
- name: init-data-migration
image: busybox
command: ['sh', '-c', "until nc -w 2 {{ .Release.Name }}-{{ index .Values.mysql.host }} {{ .Values.mysql.service.port }}; do echo Waiting for {{ .Release.Name }}-{{ index .Values.mysql.host }}; sleep 2; done;"]
containers:
- name: {{ .Chart.Name }}
image: {{ include "k8-golang-demo.migration-image" . }}
imagePullPolicy: {{ .Values.migration.image.pullPolicy }}
env:
- name: MYSQL_HOST
value: "{{ .Release.Name }}-{{ index .Values.mysql.host }}"
- name: MYSQL_PORT_NUMBER
value: {{ .Values.mysql.service.port | quote }}
- name: MYSQL_DATABASE_NAME
value: {{ .Values.mysql.mysqlDatabase | quote }}
- name: MYSQL_DATABASE_USER
value: {{ .Values.mysql.mysqlUser | quote }}
- name: MYSQL_DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
key: mysqlPassword
- name: MYSQL_DRIVER
value: {{ .Values.mysql.mysqlDriver | quote }}
В секции containers мы описали, какой контейнер стоит использовать и с какими переменными окружениями. Все значения беруться из файла values, кроме пароля, который считывается из специального ресурса - secrets.
secrets.yaml
apiVersion: v1
kind: Secret
metadata:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
namespace: {{ .Release.Namespace }}
labels:
{{- include "k8-golang-demo.labels" . | nindent 4 }}
type: Opaque
data:
mysqlPassword: {{ .Values.mysql.mysqlPassword | b64enc | quote }}
У нас появился раздел initContainers, в котором мы просто пингуем базу и проверяем, поднялась она или нет. Этот маленький трюк позволяет нам запускать миграцию данных только тогда, когда база данных запущена и готова к приему трафика.
Также стоит обратить внимание, что у джобов есть так называемые hooks. Их суть заключается в том, чтобы при наступлении одного из перечисленных событий, в нашем случае при post-install, post-upgrade Helm будет создавать джобу для миграции данных.
Helm сортирует хуки по весу (по умолчанию назначается вес 0), по типу ресурса и, наконец, по имени в порядке возрастания. В качестве веса, задано значение -5, так наши миграции будут более приоритетней других задач.
Можно определить политику, которые определяют, когда удалять соответствующие ресурсы. Политики удаления хуков определяются с помощью следующей аннотации:
skaffold.yaml
apiVersion: skaffold/v2beta26
kind: Config
build:
artifacts:
- image: docker.io/golang-enthusiast/k8-golang-demo
- image: docker.io/golang-enthusiast/k8-golang-demo-data-migration
docker:
dockerfile: ./migrations/Dockerfile
local:
push: false
concurrency: 1
deploy:
helm:
flags:
upgrade: ["--timeout", "15m"]
install: ["--timeout", "15m"]
releases:
- name: test
chartPath: helm/k8-golang-demo
wait: true
artifactOverrides:
skaffoldImage: docker.io/golang-enthusiast/k8-golang-demo
migration.skaffoldImage: docker.io/golang-enthusiast/k8-golang-demo-data-migration
setValueTemplates:
skaffold: true
image.pullPolicy: Never
migration.image.pullPolicy: Never
profiles:
- name: mysql
patches:
- op: add
path: /deploy/helm/releases/0/setValueTemplates/mysql.enabled
value: true
portForward:
- resourceType: deployment
resourceName: test-k8-golang-demo
namespace: default
port: 8080
localPort: 8080
На стадии build мы указываем, что собрать и сохранить образы нужно локально. Skaffold соберет нам два образа, один для основного приложения, а другой для миграции данных. Мы также указали, что не обязательно загружать собранные образы в репозиторий (push: false). Флаг concurrency указывает, что образы должны собираться параллельно.
На стадии deploy мы указываем, как должно деплоиться приложение в minikube. В нашем случае, через Helm - мы указывает путь до helm чарта, а также значения, которые должны быть переопределены. Например, если взведен флаг skaffold: true, то мы будем использовать образы из значений skaffoldImage и migration.skaffoldImage. Логика, что и когда использовать указана в файле:
_helpers.tpl
{{/*
Change how the image is assigned based on the skaffold flag.
*/}}
{{- define "k8-golang-demo.image" -}}
{{- if .Values.skaffold -}}
{{- .Values.skaffoldImage -}}
{{- else -}}
{{- printf "%s%s:%s" .Values.image.hostname .Values.image.repository .Values.image.tag -}}
{{- end -}}
{{- end -}}
{{/*
Change how the data migration image is assigned based on the skaffold flag.
*/}}
{{- define "k8-golang-demo.migration-image" -}}
{{- if .Values.skaffold -}}
{{- .Values.migration.skaffoldImage -}}
{{- else -}}
{{- printf "%s%s:%s" .Values.migration.image.hostname .Values.migration.image.repository .Values.migration.image.tag -}}
{{- end -}}
{{- end -}}
portForward: аналогично тому, как мы обычно прокидываем порты с помощью kubectl port-forward
profiles - профили запуска, например, если приложение может работать с несколькими базами данными, mysql или postgres, то мы можем указать несколько профилей, по одному на каждую базу данных.
$ minikube start
Так как, мы используем bitnami в качестве зависимости для базы данных MySQL, то нужно добавить данный репозиторий в Helm:
$ helm repo add bitnami https://charts.bitnami.com/bitnami
Проверим добавленные репозитории командой:
$ helm repo list
NAME URL
bitnami https://charts.bitnami.com/bitnami
Обновляем Helm зависимости, командой:
$ helm dep up ./helm/k8-golang-demo/
Запускаем skaffold:
$ skaffold run -p mysql --port-forward=user --no-prune=false --cache-artifacts=false
Подождем пару минут и приложение должно успешно задеплоиться в minikube. Весь исходный код доступен на GitHub.
habr.com
Простейший сервис на Golang
Итак, у нас есть очень простое приложение на Golang, которое использует как gRPC, так и протокол REST в качестве транспортного протокола. У нашего сервиса только два ендпоинта, которые записывают и считывают данные из базы данных MySQL.Хочу заранее отметить, что мое приложение носит исключительно демонстрационный характер, поэтому весь исходный код в одном main файле:
cmd/app/main.go
package main
import (
"context"
"database/sql"
"errors"
"flag"
"fmt"
"log"
"net"
"net/http"
"os"
"os/signal"
"syscall"
"time"
apiv1 "k8-golang-demo/gen/pb-go/com.example/usersvcapi/v1"
"github.com/google/uuid"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
run "github.com/oklog/oklog/pkg/group"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
// justifying it
_ "github.com/go-sql-driver/mysql"
)
func main() {
// Flags.
//
fs := flag.NewFlagSet("", flag.ExitOnError)
grpcAddr := fs.String("grpc-addr", ":6565", "grpc address")
httpAddr := fs.String("http-addr", ":8080", "http address")
if err := fs.Parse(os.Args[1:]); err != nil {
log.Fatal(err)
}
// Setup database.
//
db, err := NewDatabase(
os.Getenv("DATABASE_DRIVER"),
os.Getenv("DATABASE_NAME"),
os.Getenv("DATABASE_USERNAME"),
os.Getenv("DATABASE_PASSWORD"),
os.Getenv("DATABASE_HOST"),
os.Getenv("DATABASE_PORT"),
)
if err != nil {
log.Fatal(err)
}
conn := db.GetConnection()
defer func() {
_ = db.CloseConnection()
}()
// Setup gRPC servers.
//
baseGrpcServer := grpc.NewServer()
userGrpcServer := NewUserGRPCServer(conn, "users")
apiv1.RegisterUserServiceServer(baseGrpcServer, userGrpcServer)
// Setup gRPC gateway.
//
ctx := context.Background()
rmux := runtime.NewServeMux()
mux := http.NewServeMux()
mux.Handle("/", rmux)
{
err := apiv1.RegisterUserServiceHandlerServer(ctx, rmux, userGrpcServer)
if err != nil {
log.Fatal(err)
}
}
// Serve.
//
var g run.Group
{
grpcListener, err := net.Listen("tcp", *grpcAddr)
if err != nil {
log.Fatal(err)
}
g.Add(func() error {
log.Printf("Serving grpc address %s", *grpcAddr)
return baseGrpcServer.Serve(grpcListener)
}, func(error) {
grpcListener.Close()
})
}
{
httpListener, err := net.Listen("tcp", *httpAddr)
if err != nil {
log.Fatal(err)
}
g.Add(func() error {
log.Printf("Serving http address %s", *httpAddr)
return http.Serve(httpListener, mux)
}, func(err error) {
httpListener.Close()
})
}
{
cancelInterrupt := make(chan struct{})
g.Add(func() error {
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
select {
case sig := <-c:
return fmt.Errorf("received signal %s", sig)
case <-cancelInterrupt:
return nil
}
}, func(error) {
close(cancelInterrupt)
})
}
if err := g.Run(); err != nil {
log.Fatal(err)
}
}
type userServer struct {
conn *sql.DB
tableName string
}
func NewUserGRPCServer(conn *sql.DB, tableName string) apiv1.UserServiceServer {
return &userServer{
conn: conn,
tableName: tableName,
}
}
func (s *userServer) CreateUser(ctx context.Context, req *apiv1.CreateUserRequest) (*apiv1.CreateUserResponse, error) {
if req.User == nil {
return nil,
status.Error(codes.InvalidArgument, "User required")
}
id, err := uuid.NewRandom()
if err != nil {
return nil,
status.Error(codes.Internal, err.Error())
}
query := `INSERT INTO ` + s.tableName + `(id, name, type) VALUES(?,?,?)`
_, err = runWriteTransaction(s.conn, query, id, req.User.Name, req.User.Type)
if err != nil {
return nil, err
}
return &apiv1.CreateUserResponse{
Id: id.String(),
}, nil
}
func (s *userServer) GetUsers(ctx context.Context, req *apiv1.GetUsersRequest) (*apiv1.GetUsersResponse, error) {
query := `SELECT id, name, type FROM ` + s.tableName + ``
users := []*apiv1.UserRead{}
err := runQuery(s.conn, query, []interface{}{}, func(rows *sql.Rows) error {
found := apiv1.UserRead{}
err := rows.Scan(
&found.Id,
&found.Name,
&found.Type,
)
if err != nil {
return err
}
users = append(users, &found)
return nil
})
if err != nil {
return nil,
status.Error(codes.NotFound, fmt.Errorf("User not found, err %v", err).Error())
}
return &apiv1.GetUsersResponse{
Users: users,
}, nil
}
// SQLDatabase is the interface that provides sql methods.
type SQLDatabase interface {
GetConnection() *sql.DB
CloseConnection() error
}
type db struct {
conn *sql.DB
}
// NewDatabase creates a new sql database connection with the base migration setup.
func NewDatabase(driver, database, username, password, host string, port string) (SQLDatabase, error) {
source := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", username, password, host, port, database)
sqlconn, err := sql.Open(driver, source)
if err != nil {
return nil, fmt.Errorf("Failed to open database connection, err:%v", err)
}
return &db{
conn: sqlconn,
}, nil
}
func (h *db) GetConnection() *sql.DB {
return h.conn
}
func (h *db) CloseConnection() error {
if h.conn == nil {
return errors.New("Cannot close the connection because the connection is nil")
}
return h.conn.Close()
}
func runWriteTransaction(db *sql.DB, query string, args ...interface{}) (sql.Result, error) {
ctx, stop := context.WithCancel(context.Background())
defer stop()
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelDefault})
if err != nil {
return nil, err
}
defer func() {
_ = tx.Rollback() // The rollback will be ignored if the tx has been committed later in the function.
}()
stmt, err := tx.Prepare(query)
if err != nil {
return nil, err
}
defer stmt.Close() // Prepared statements take up server resources and should be closed after use.
queryResult, err := stmt.Exec(args...)
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return queryResult, err
}
func runQuery(db *sql.DB, query string, args []interface{}, f func(*sql.Rows) error) error {
ctx, stop := context.WithCancel(context.Background())
defer stop()
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
queryResult, err := db.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer func() {
_ = queryResult.Close()
}()
for queryResult.Next() {
err = f(queryResult)
if err != nil {
return err
}
}
return nil
}
Чтобы приложение было готово к запуску в Kubernetes, нам необходимо контейнеризировать его. Опишем докер файл для нашего сервиса:
Dockerfile
FROM golang:1.16-alpine as builder
WORKDIR /build
COPY go.mod .
COPY go.sum .
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o /main cmd/app/main.go
FROM alpine:3
COPY --from=builder main /bin/main
ENTRYPOINT ["/bin/main"]
Миграция данных
Если прямо сейчас мы попробуем сконфигурировать бд и запустить приложение, то оно упадет, т.к. схема для таблицы users не готова. Для того, чтобы создать нужные схемы, нам нужны миграции данных.Опишем код для миграции данных, которые на входе будут принимать SQL файлы и выполнять их в соответствующем порядке. В качестве фреймворка мы будем использовать достаточно популярный инструмент golang-migrate:
cmd/migrations/main.go
package main
import (
"fmt"
"os"
"database/sql"
"github.com/cenk/backoff"
_ "github.com/go-sql-driver/mysql"
"github.com/golang-migrate/migrate/v4"
"github.com/golang-migrate/migrate/v4/database/mysql"
_ "github.com/golang-migrate/migrate/v4/source/file"
)
const migrationsPath = "file://migrations"
func main() {
// Open connection.
conn, driver := getConnection()
// Close connection.
defer func() {
if conn != nil {
_ = conn.Close()
}
}()
// Establish connection to a database.
establishConnectionWithRetry(conn)
// Starting migration job.
err := migrateSQL(conn, driver)
if err != nil {
panic(err)
} else {
fmt.Println("Migration successfully finished.")
}
}
func getConnection() (*sql.DB, string) {
driver := os.Getenv("MYSQL_DRIVER")
address := os.Getenv("MYSQL_HOST") + ":" + os.Getenv("MYSQL_PORT_NUMBER")
username := os.Getenv("MYSQL_DATABASE_USER")
password := os.Getenv("MYSQL_DATABASE_PASSWORD")
database := os.Getenv("MYSQL_DATABASE_NAME")
// Open may just validate its arguments without creating a connection to the database.
sqlconn, err := sql.Open(driver, username+":"+password+"@tcp("+address+")/"+database)
if err != nil {
panic("Cannot establish connection to a database")
}
return sqlconn, driver
}
// This function executes the migration scripts.
func migrateSQL(conn *sql.DB, driverName string) error {
driver, _ := mysql.WithInstance(conn, &mysql.Config{})
m, err := migrate.NewWithDatabaseInstance(
migrationsPath,
driverName,
driver,
)
if err != nil {
return err
}
if err := m.Up(); err != nil && err != migrate.ErrNoChange {
return err
}
return nil
}
func establishConnectionWithRetry(conn *sql.DB) {
b := backoff.NewExponentialBackOff()
// We wait forever until the connection will be established.
// In practice k8s will kill the pod if it takes too long.
b.MaxElapsedTime = 0
_ = backoff.Retry(func() error {
fmt.Println("Connecting to a database ...")
// Ping verifies a connection to the database is still alive,
// establishing a connection if necessary.
if errPing := conn.Ping(); errPing != nil {
return fmt.Errorf("ping failed %v", errPing)
}
return nil
}, b)
}
Теперь когда утилита для миграции данных готова, нам нужно описать сами миграции, для этого в директории migrations/sql создадим:
1_create_table_foobar.up.sql
CREATE TABLE IF NOT EXISTS users (
id VARCHAR(40),
name VARCHAR(255),
type SMALLINT UNSIGNED,
PRIMARY KEY(id)
);
В кластере Kubernetes мы будем запускать миграции данных в качестве джобов, и соответственно нам их также нужно контейнеризировать:
migrations/Dockerfile
FROM golang:1.16-alpine as builder
WORKDIR /build
COPY go.mod .
COPY go.sum .
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o /main cmd/migrations/main.go
FROM alpine:3
COPY --from=builder main /bin/main
COPY --from=builder build/migrations/sql /migrations
ENTRYPOINT ["/bin/main"]
Мы только что создали контейнер для миграции данных и скопировали туда SQL файлы. Наше приложение почти готово к запуску в Kubernetes, самое время описать манифесты.
Kubernetes
Прежде чем мы начнем, нам понадобятся следующие инструменты:- minikube мы будем использовать в качестве тестовой среды, очень удобный инструмент, который позволяет развернуть всё необходимое на локальной машине.
- helm charts поможет нам упаковать наше приложение для установки в кластеры Kubernetes.
- skaffold дает возможность быстро получать результат очередных изменений кода — в виде обновлённого приложения, работающего в кластере Kubernetes.
Helm charts
Когда все инструменты установлены, мы можем приступить к описанию k8s манифестов с помощью helm charts. Для начала создадим базовый шаблон для манифестов, а затем отредактируем его под свои нужды. В директории helm запускаем следующую команду:helm create k8-golang-demo
Отлично, только что мы создали базовый шаблон с определениями ресурсов, необходимые для запуска приложения внутри кластера Kubernetes. Приступим к редактированию ресурсов шаблона под наши нужды.
Chart.yaml - содержит метаданные, общую информацию о чарте, а так же зависимости на другие helm чарты. Поскольку наше приложение использует базу данных MySQL в качестве зависимости, нам необходимо описать эту зависимость в блоке dependencies:
Chart.yaml
Из важного:
- name - имя нашего чарта
- version и appVersion - версия чарта и версия приложения. Обе версии могут совпадать или иметь отдельные версии. Это больше зависит от ваших предпочтений.
- В блоке зависимостей мы описали зависимость от базы данных MySQL, указав версию и репозиторий helm чарта. Мы можем указать псевдоним (alias) для нашей зависимости, а также условие (condition), по которому стоит включать или не включать эту зависимость. Например, в разработке мы хотим развернуть базу данных, а в продакшене мы хотим использовать облачную базу данных, и поэтому мы должны отключить эту зависимость в продакшене.
values.yaml
# Default values for k8-golang-demo.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.
replicaCount: 1
image:
hostname: docker.io
repository: /golang-enthusiast/k8-golang-demo
tag: 0.1.0
pullPolicy: IfNotPresent
migration:
image:
hostname: docker.io
repository: /golang-enthusiast/k8-golang-demo-data-migration
tag: 0.3.0
pullPolicy: IfNotPresent
service:
type: ClusterIP
protocol: TCP
port: 6565
httpPort: 8080
name: grpc
ingress:
name: http
protocol: HTTP
port: 80
extension: svc.cluster.local
serviceAccount:
create: true
mysql:
enabled: false
host: mysql
mysqlDriver: mysql
mysqlRootPassword: test
mysqlDatabase: test
mysqlUser: admin
mysqlPassword: test
service:
port: 3306
initdbScripts:
initdb.sql: |-
CREATE DATABASE IF NOT EXISTS test DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;
CREATE USER 'admin'@'%' IDENTIFIED BY 'test';
GRANT ALL PRIVILEGES ON *.* TO 'admin'@'%' WITH GRANT OPTION;
FLUSH PRIVILEGES;
primary:
persistence:
enabled: false
storageClass: standard
grpcGateway:
service:
protocol: TCP
port: 8080
name: http
skaffold: false
В значениях мы описали конфигурацию базы данных MySQL. В общем, конфигурация может различаться от поставщика к поставщику helm чартов. Т.к. мы используем стабильные чарты от bitnami, то и конфигурацию мы должны делать соответствующую, которая описана тут.
Так же мы описали образы, которые будут использоваться в нашем приложении. Их всего два, а именно образ для основного приложения и образ для миграции данных. К остальным значениям мы вернемся чуть позже.
deployment.yaml - содержит инструкции для развертывания приложения.
deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ include "k8-golang-demo.fullname" . }}
namespace: {{ .Release.Namespace }}
labels:
{{- include "k8-golang-demo.labels" . | nindent 4 }}
spec:
replicas: {{ .Values.replicaCount }}
selector:
matchLabels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
template:
metadata:
labels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
spec:
containers:
- name: {{ .Chart.Name }}
image: {{ include "k8-golang-demo.image" . }}
imagePullPolicy: {{ .Values.image.pullPolicy }}
env:
- name: DATABASE_HOST
value: "{{ .Release.Name }}-{{ index .Values.mysql.host }}"
- name: DATABASE_PORT
value: {{ .Values.mysql.service.port | quote }}
- name: DATABASE_NAME
value: {{ .Values.mysql.mysqlDatabase | quote }}
- name: DATABASE_USERNAME
value: {{ .Values.mysql.mysqlUser | quote }}
- name: DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
key: mysqlPassword
- name: DATABASE_DRIVER
value: {{ .Values.mysql.mysqlDriver | quote }}
ports:
- name: {{ .Values.service.name }}
containerPort: {{ .Values.service.port }}
protocol: {{ .Values.service.protocol }}
- name: {{ .Values.grpcGateway.service.name }}
containerPort: {{ .Values.grpcGateway.service.port }}
protocol: {{ .Values.grpcGateway.service.protocol }}
В инструкции по развертыванию приложения в секции containers мы описали какой контейнер стоит использовать и с какими переменными окружениями, так же указали какие порты должны быть использованы. Т.к. наше приложение использует два транспортных протокола - gRPC + REST, то следует указать описание обоих в секции ports.
data-migration-job.yaml - здесь мы будем описывать инструкции по развертыванию заданий (джобов) для миграции данных.
data-migration-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
name: {{ include "k8-golang-demo.fullname" . }}-data-migration-job
namespace: {{ .Release.Namespace }}
labels:
{{ include "k8-golang-demo.labels" . | indent 4 }}
annotations:
# This is what defines this resource as a hook. Without this line, the
# job is considered part of the release.
"helm.sh/hook": post-install, post-upgrade
"helm.sh/hook-weight": "-5"
"helm.sh/hook-delete-policy": before-hook-creation
spec:
template:
metadata:
labels:
app.kubernetes.io/name: {{ include "k8-golang-demo.name" . }}-migration
app.kubernetes.io/instance: {{ .Release.Name }}
annotations:
readiness.status.sidecar.istio.io/applicationPorts: ""
spec:
restartPolicy: Never
initContainers:
- name: init-data-migration
image: busybox
command: ['sh', '-c', "until nc -w 2 {{ .Release.Name }}-{{ index .Values.mysql.host }} {{ .Values.mysql.service.port }}; do echo Waiting for {{ .Release.Name }}-{{ index .Values.mysql.host }}; sleep 2; done;"]
containers:
- name: {{ .Chart.Name }}
image: {{ include "k8-golang-demo.migration-image" . }}
imagePullPolicy: {{ .Values.migration.image.pullPolicy }}
env:
- name: MYSQL_HOST
value: "{{ .Release.Name }}-{{ index .Values.mysql.host }}"
- name: MYSQL_PORT_NUMBER
value: {{ .Values.mysql.service.port | quote }}
- name: MYSQL_DATABASE_NAME
value: {{ .Values.mysql.mysqlDatabase | quote }}
- name: MYSQL_DATABASE_USER
value: {{ .Values.mysql.mysqlUser | quote }}
- name: MYSQL_DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
key: mysqlPassword
- name: MYSQL_DRIVER
value: {{ .Values.mysql.mysqlDriver | quote }}
В секции containers мы описали, какой контейнер стоит использовать и с какими переменными окружениями. Все значения беруться из файла values, кроме пароля, который считывается из специального ресурса - secrets.
secrets.yaml
apiVersion: v1
kind: Secret
metadata:
name: {{ include "k8-golang-demo.fullname" . }}-secrets
namespace: {{ .Release.Namespace }}
labels:
{{- include "k8-golang-demo.labels" . | nindent 4 }}
type: Opaque
data:
mysqlPassword: {{ .Values.mysql.mysqlPassword | b64enc | quote }}
У нас появился раздел initContainers, в котором мы просто пингуем базу и проверяем, поднялась она или нет. Этот маленький трюк позволяет нам запускать миграцию данных только тогда, когда база данных запущена и готова к приему трафика.
Также стоит обратить внимание, что у джобов есть так называемые hooks. Их суть заключается в том, чтобы при наступлении одного из перечисленных событий, в нашем случае при post-install, post-upgrade Helm будет создавать джобу для миграции данных.
Helm сортирует хуки по весу (по умолчанию назначается вес 0), по типу ресурса и, наконец, по имени в порядке возрастания. В качестве веса, задано значение -5, так наши миграции будут более приоритетней других задач.
Можно определить политику, которые определяют, когда удалять соответствующие ресурсы. Политики удаления хуков определяются с помощью следующей аннотации:
| before-hook-creation | Delete the previous resource before a new hook is launched (default) |
| hook-succeeded | Delete the resource after the hook is successfully executed |
| hook-failed | Delete the resource if the hook failed during execution |
Skaffold
Отличный инструмент для разработки и отладки приложений в Kubernetes, более подробно можно почитать тут. Что лично мне понравилось:- Позволяет в фоновом режиме следить за изменениями в исходном коде и запускать автоматизированный процесс сборки кода в образы контейнеров, деплой этих образов в кластер Kubernetes
- Синхронизирует файлы в репозитории с рабочим каталогом в контейнере
- Автоматически пробрасывает порты и читает логи приложения, запущенного в контейнере
skaffold.yaml
apiVersion: skaffold/v2beta26
kind: Config
build:
artifacts:
- image: docker.io/golang-enthusiast/k8-golang-demo
- image: docker.io/golang-enthusiast/k8-golang-demo-data-migration
docker:
dockerfile: ./migrations/Dockerfile
local:
push: false
concurrency: 1
deploy:
helm:
flags:
upgrade: ["--timeout", "15m"]
install: ["--timeout", "15m"]
releases:
- name: test
chartPath: helm/k8-golang-demo
wait: true
artifactOverrides:
skaffoldImage: docker.io/golang-enthusiast/k8-golang-demo
migration.skaffoldImage: docker.io/golang-enthusiast/k8-golang-demo-data-migration
setValueTemplates:
skaffold: true
image.pullPolicy: Never
migration.image.pullPolicy: Never
profiles:
- name: mysql
patches:
- op: add
path: /deploy/helm/releases/0/setValueTemplates/mysql.enabled
value: true
portForward:
- resourceType: deployment
resourceName: test-k8-golang-demo
namespace: default
port: 8080
localPort: 8080
На стадии build мы указываем, что собрать и сохранить образы нужно локально. Skaffold соберет нам два образа, один для основного приложения, а другой для миграции данных. Мы также указали, что не обязательно загружать собранные образы в репозиторий (push: false). Флаг concurrency указывает, что образы должны собираться параллельно.
На стадии deploy мы указываем, как должно деплоиться приложение в minikube. В нашем случае, через Helm - мы указывает путь до helm чарта, а также значения, которые должны быть переопределены. Например, если взведен флаг skaffold: true, то мы будем использовать образы из значений skaffoldImage и migration.skaffoldImage. Логика, что и когда использовать указана в файле:
_helpers.tpl
{{/*
Change how the image is assigned based on the skaffold flag.
*/}}
{{- define "k8-golang-demo.image" -}}
{{- if .Values.skaffold -}}
{{- .Values.skaffoldImage -}}
{{- else -}}
{{- printf "%s%s:%s" .Values.image.hostname .Values.image.repository .Values.image.tag -}}
{{- end -}}
{{- end -}}
{{/*
Change how the data migration image is assigned based on the skaffold flag.
*/}}
{{- define "k8-golang-demo.migration-image" -}}
{{- if .Values.skaffold -}}
{{- .Values.migration.skaffoldImage -}}
{{- else -}}
{{- printf "%s%s:%s" .Values.migration.image.hostname .Values.migration.image.repository .Values.migration.image.tag -}}
{{- end -}}
{{- end -}}
portForward: аналогично тому, как мы обычно прокидываем порты с помощью kubectl port-forward
profiles - профили запуска, например, если приложение может работать с несколькими базами данными, mysql или postgres, то мы можем указать несколько профилей, по одному на каждую базу данных.
Запуск приложения
Для начала нам нужно запустить minikube, командой:$ minikube start
Так как, мы используем bitnami в качестве зависимости для базы данных MySQL, то нужно добавить данный репозиторий в Helm:
$ helm repo add bitnami https://charts.bitnami.com/bitnami
Проверим добавленные репозитории командой:
$ helm repo list
NAME URL
bitnami https://charts.bitnami.com/bitnami
Обновляем Helm зависимости, командой:
$ helm dep up ./helm/k8-golang-demo/
Запускаем skaffold:
$ skaffold run -p mysql --port-forward=user --no-prune=false --cache-artifacts=false
Подождем пару минут и приложение должно успешно задеплоиться в minikube. Весь исходный код доступен на GitHub.
Разворачиваем Golang приложение в Kubernetes
В этой статье я хочу поделиться примером, как можно развернуть простое приложение на Golang в Kubernetes, с помощью helm чартов и skaffold скриптов. Думаю, данная статья может быть полезной тем...
habr.com